[SPARK-24257][SQL]LongToUnsafeRowMap calculate the new size may be wrong #21311

cxzl25 · 2018-05-12T11:18:27Z

What changes were proposed in this pull request?

LongToUnsafeRowMap
Calculate the new size simply by multiplying by 2
At this time, the size of the application may not be enough to store data
Some data is lost and the data read out is dirty

How was this patch tested?

HashedRelationSuite
test("LongToUnsafeRowMap with big values")

maropu · 2018-05-14T02:07:05Z

@gatorsmile @hvanhovell Could you trigger tests?

kiszk · 2018-05-14T07:28:26Z

cc @cloud-fan

kiszk · 2018-05-14T07:35:03Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala

-      ensureAcquireMemory(used * 8L * 2)
-      val newPage = new Array[Long](used * 2)
+      val multiples = math.max(math.ceil(needSize.toDouble / (used * 8L)).toInt, 2)
+      ensureAcquireMemory(used * 8L * multiples)


Do we move the size check into before ensureAcquireMemory()? IIUC, we have to check used * multiplies <= ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH` now.

How about shaping up this logic along with the other similar ones (spliting this func into two parts: grow/append)? e.g., UTF8StringBuilder https://github.com/apache/spark/blob/master/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UTF8StringBuilder.java#L43

+1 on grow/append

ok.Spliting append func into two parts: grow/append.

maropu · 2018-05-14T07:51:41Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala

-    if (cursor + 8 + row.getSizeInBytes > page.length * 8L + Platform.LONG_ARRAY_OFFSET) {
+    val needSize = cursor + 8 + row.getSizeInBytes
+    val nowSize = page.length * 8L + Platform.LONG_ARRAY_OFFSET
+    if (needSize > nowSize) {
      val used = page.length
      if (used >= (1 << 30)) {
        sys.error("Can not build a HashedRelation that is larger than 8G")


This is not related to this pr though, sys.error instead of UnsupportedOperationException?

spark/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UTF8StringBuilder.java

Line 45 in b6c50d7

throw new UnsupportedOperationException(

ok. sys.error instead of UnsupportedOperationException

cloud-fan · 2018-05-14T09:00:04Z

ok to test

cloud-fan · 2018-05-14T09:16:51Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala

      val used = page.length
      if (used >= (1 << 30)) {
        sys.error("Can not build a HashedRelation that is larger than 8G")
      }
-      ensureAcquireMemory(used * 8L * 2)


Doubling the size when growing is very typical, seems what you want to address is when the memory is enough for the requsted size but not enough for doubling the size. I'd suggest we should double the size most of the time, as long as there is enough memory.

ok . Doubling the size when growing.

…n growing;sys.error instead of UnsupportedOperationException

cxzl25 · 2018-05-14T10:48:37Z

Thanks for your review. @maropu @kiszk @cloud-fan

I submitted a modification including the following:

spliting append func into two parts:grow/appendG
doubling the size when growing
sys.error instead of UnsupportedOperationException

SparkQA · 2018-05-14T16:13:17Z

Test build #90575 has finished for PR 21311 at commit d9d8e62.

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

SparkQA · 2018-05-14T16:22:37Z

Test build #90574 has finished for PR 21311 at commit 22a2767.

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

JoshRosen · 2018-05-22T08:46:38Z

@cxzl25, to clarify:

Some data is lost and the data read out is dirty

To clarify, is this a potential cause of a wrong-answer correctness bug? If so, we should be sure to backport the resulting fix to maintenance branches. /cc @cloud-fan @gatorsmile

cxzl25 · 2018-05-22T09:28:16Z

@JoshRosen @cloud-fan @gatorsmile
When introducing SPARK-10399,UnsafeRow#getUTF8String check the size at this time.
UnsafeRow#getUTF8String
OnHeapMemoryBlock

The sum of size 2097152 and offset 32 should not be larger than the size of the given memory space 2097168

But when this patch is not introduced, no error, get wrong value.

cloud-fan · 2018-05-22T11:33:06Z

Calculate the new size simply by multiplying by 2
At this time, the size of the application may not be enough to store data
Some data is lost and the data read out is dirty

Can you explain more about it? IIUC if we don't have enough memory for size * 2, we would just fail with OOM, instead of setting a wrong size.

cxzl25 · 2018-05-22T11:47:43Z

@cloud-fan
LongToUnsafeRowMap#append(key: Long, row: UnsafeRow)
when row.getSizeInBytes > newPageSize( oldPage.length * 8L * 2),still use newPageSize value.
When the new page size is insufficient to hold the entire row of data, Platform.copyMemory is still called.No error.
At this time, the last remaining data was discarded.
When reading data, read this buffer according to offset and length,the last data is unpredictable.

cloud-fan · 2018-05-22T12:17:21Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala

@@ -626,6 +618,32 @@ private[execution] final class LongToUnsafeRowMap(val mm: TaskMemoryManager, cap
    }
  }

+  private def grow(neededSize: Int): Unit = {
+    // There is 8 bytes for the pointer to next value
+    val totalNeededSize = cursor + 8 + neededSize


The grow logic should be: we must grow to fit the new row, otherwise OOM should be thrown. If possible, grow to oldSize * 2

private def grow(inputRowSize: Int): Unit = { val neededNumWords = (cursor - Platform.LONG_ARRAY_OFFSET + 8 + inputRowSize + 7) / 8 if (neededNumWords > page.length) { if (neededNumWords > (1 << 30)) fail... val newNumWords = math.max(neededNumWords, math.min(page.length * 2, 1 << 30)) ensureAcquireMemory(newNumWords * 8L) ... } }

@cloud-fan Thank you for your suggestion and code.

…grow to oldSize * 2

cloud-fan · 2018-05-22T13:51:19Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala

@@ -626,6 +618,29 @@ private[execution] final class LongToUnsafeRowMap(val mm: TaskMemoryManager, cap
    }
  }

+  private def grow(inputRowSize: Int): Unit = {
+    val neededNumWords = (cursor - Platform.LONG_ARRAY_OFFSET + 8 + inputRowSize + 7) / 8


don't forget the comment for the 8 bytes pointer

cloud-fan · 2018-05-22T13:52:34Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala

+          "Can not build a HashedRelation that is larger than 8G")
+      }
+      val newNumWords = math.max(neededNumWords, math.min(page.length * 2, 1 << 30))
+      if (newNumWords > ARRAY_MAX) {


we won't need this check now, newNumWords is guaranteed to be less than (1 << 30), which is much smaller than ARRAY_MAX

cloud-fan · 2018-05-22T13:58:20Z

sql/core/src/test/scala/org/apache/spark/sql/execution/joins/HashedRelationSuite.scala

+    val unsafeProj = UnsafeProjection.create(Seq(BoundReference(0, StringType, false)))
+    val keys = Seq(0L)
+    val map = new LongToUnsafeRowMap(taskMemoryManager, 1)
+    val bigStr = UTF8String.fromString("x" * 1024 * 1024 * 2)


let's add a comment to say, the page array is initialized with length 1 << 17, so here we need a value larger than 1 << 18, to trigger the bug

cloud-fan · 2018-05-22T13:58:44Z

sql/core/src/test/scala/org/apache/spark/sql/execution/joins/HashedRelationSuite.scala

+    val keys = Seq(0L)
+    val map = new LongToUnsafeRowMap(taskMemoryManager, 1)
+    val bigStr = UTF8String.fromString("x" * 1024 * 1024 * 2)
+    keys.foreach { k =>


we just have one key, why use loop?

cloud-fan · 2018-05-22T13:59:02Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/HashedRelation.scala

@@ -30,6 +30,7 @@ import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.plans.physical.BroadcastMode
 import org.apache.spark.sql.types.LongType
 import org.apache.spark.unsafe.Platform
+import org.apache.spark.unsafe.array.ByteArrayMethods


cloud-fan · 2018-05-22T14:01:17Z

sql/core/src/test/scala/org/apache/spark/sql/execution/joins/HashedRelationSuite.scala

+        Long.MaxValue,
+        1),
+      0)
+    val unsafeProj = UnsafeProjection.create(Seq(BoundReference(0, StringType, false)))


nit: UnsafeProjection.create(Array(StringType))

cloud-fan · 2018-05-22T14:02:46Z

sql/core/src/test/scala/org/apache/spark/sql/execution/joins/HashedRelationSuite.scala

+      map.append(k, unsafeProj(InternalRow(bigStr)))
+    }
+    map.optimize()
+    val row = unsafeProj(InternalRow(bigStr)).copy()


val resultRow = new UnsafeRow(1)

cloud-fan · 2018-05-22T15:04:26Z

sql/core/src/test/scala/org/apache/spark/sql/execution/joins/HashedRelationSuite.scala

+    val key = 0L
+    // the page array is initialized with length 1 << 17,
+    // so here we need a value larger than 1 << 18
+    val bigStr = UTF8String.fromString("x" * 1024 * 1024 * 2)


nit: can we just do "x" * (1 << 19) here?

cloud-fan · 2018-05-22T15:04:55Z

LGTM, good catch!

SparkQA · 2018-05-22T15:49:51Z

Test build #90966 has finished for PR 21311 at commit 6fe1dd0.

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

cloud-fan · 2018-05-22T16:48:14Z

sql/core/src/test/scala/org/apache/spark/sql/execution/joins/HashedRelationSuite.scala

+    val key = 0L
+    // the page array is initialized with length 1 << 17 (1M bytes),
+    // so here we need a value larger than 1 << 18 (2M bytes),to trigger the bug
+    val bigStr = UTF8String.fromString("x" * (1 << 22))


to double check, do we have to use 1 << 22 to trigger this bug?

Not necessary.
Just chose a larger value to make it easier to lose data.

do you mean this bug can't be reproduced consistently? e.g. if we pick 1 << 18 + 1, we may not expose this bug, so we have to use 1 << 22 to 100% reproduce this bug?

LongToUnsafeRowMap#getRow
resultRow=UnsafeRow#pointTo(page(1<<18), baseOffset(16), sizeInBytes(1<<21+16))

UTF8String#getBytes
copyMemory(base(page), offset, bytes, BYTE_ARRAY_OFFSET, numBytes(1<<21+16));

In the case of similar size sometimes, can still read the original value.

When introducing SPARK-10399,UnsafeRow#getUTF8String check the size at this time.
If we pick 1 << 18 + 1, 100% reproduce this bug.

But when this patch is not introduced, differences that are too small sometimes do not trigger.
So I chose a larger value.

My understanding may be problematic. Please advise. Thank you.

sun.misc.Unsafe unsafe; try { Field unsafeField = Unsafe.class.getDeclaredField("theUnsafe"); unsafeField.setAccessible(true); unsafe = (sun.misc.Unsafe) unsafeField.get(null); } catch (Throwable cause) { unsafe = null; } String value = "xxxxx"; byte[] src = value.getBytes(); byte[] dst = new byte[3]; byte[] newDst = new byte[5]; unsafe.copyMemory(src, 16, dst, 16, src.length); unsafe.copyMemory(dst, 16, newDst, 16, src.length); System.out.println("dst:" + new String(dst)); System.out.println("newDst:" + new String(newDst));

output:

dst:xxx
newDst:xxxxx

then 1 << 19 should be good enough as it doubles the size?

Yes. I think so.

SparkQA · 2018-05-22T17:40:06Z

Test build #90967 has finished for PR 21311 at commit f3916e7.

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

SparkQA · 2018-05-22T18:08:51Z

Test build #90970 has finished for PR 21311 at commit d7da8ae.

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

SparkQA · 2018-05-22T18:14:52Z

Test build #90981 has finished for PR 21311 at commit b8b6324.

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

maropu · 2018-05-23T02:12:06Z

retest this please

SparkQA · 2018-05-23T05:05:17Z

Test build #91009 has finished for PR 21311 at commit b8b6324.

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

kiszk · 2018-05-23T07:00:14Z

retest this please

SparkQA · 2018-05-23T07:52:02Z

Test build #91018 has finished for PR 21311 at commit b8b6324.

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

kiszk · 2018-05-23T13:58:45Z

retest this please

SparkQA · 2018-05-23T16:37:11Z

Test build #91041 has finished for PR 21311 at commit b8b6324.

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

vanzin · 2018-05-23T17:17:48Z

retest this please

SparkQA · 2018-05-23T19:57:23Z

Test build #91052 has finished for PR 21311 at commit b8b6324.

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

gatorsmile · 2018-05-23T20:15:46Z

retest this please

SparkQA · 2018-05-24T00:02:54Z

Test build #91066 has finished for PR 21311 at commit b8b6324.

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

…rong LongToUnsafeRowMap has a mistake when growing its page array: it blindly grows to `oldSize * 2`, while the new record may be larger than `oldSize * 2`. Then we may have a malformed UnsafeRow when querying this map, whose actual data is smaller than its declared size, and the data is corrupted. Author: sychen <[email protected]> Closes #21311 from cxzl25/fix_LongToUnsafeRowMap_page_size. (cherry picked from commit 8883401) Signed-off-by: Wenchen Fan <[email protected]>

cloud-fan · 2018-05-24T03:26:28Z

thanks, merging to master/2.3/2.2/2.1/2.0! There is no conflict so I backported all the way to 2.0. I'll watch the jenkins build in the next few days.

cxzl25 · 2018-05-24T03:40:26Z

@cloud-fan Thank you very much for your help.

…rong LongToUnsafeRowMap has a mistake when growing its page array: it blindly grows to `oldSize * 2`, while the new record may be larger than `oldSize * 2`. Then we may have a malformed UnsafeRow when querying this map, whose actual data is smaller than its declared size, and the data is corrupted. Author: sychen <[email protected]> Closes apache#21311 from cxzl25/fix_LongToUnsafeRowMap_page_size. (cherry picked from commit 8883401) Signed-off-by: Wenchen Fan <[email protected]>

…rong LongToUnsafeRowMap has a mistake when growing its page array: it blindly grows to `oldSize * 2`, while the new record may be larger than `oldSize * 2`. Then we may have a malformed UnsafeRow when querying this map, whose actual data is smaller than its declared size, and the data is corrupted. Author: sychen <[email protected]> Closes apache#21311 from cxzl25/fix_LongToUnsafeRowMap_page_size.

LongToUnsafeRowMap Calculate the new correct size

d9d8e62

kiszk reviewed May 14, 2018

View reviewed changes

maropu reviewed May 14, 2018

View reviewed changes

cloud-fan reviewed May 14, 2018

View reviewed changes

spliting append func into two parts:grow/append;doubling the size whe…

22a2767

…n growing;sys.error instead of UnsupportedOperationException

cloud-fan reviewed May 22, 2018

View reviewed changes

grow to fit the new row, otherwise OOM should be thrown.If possible, …

6fe1dd0

…grow to oldSize * 2

cloud-fan reviewed May 22, 2018

View reviewed changes

Remove unnecessary checks and add comments

f3916e7

cloud-fan reviewed May 22, 2018

View reviewed changes

Remove unnecessary import and simplify test case

d7da8ae

cloud-fan reviewed May 22, 2018

View reviewed changes

add a comment

b8b6324

cloud-fan reviewed May 22, 2018

View reviewed changes

asfgit closed this in 8883401 May 24, 2018

[SPARK-24257][SQL]LongToUnsafeRowMap calculate the new size may be wrong #21311

[SPARK-24257][SQL]LongToUnsafeRowMap calculate the new size may be wrong #21311

Conversation

cxzl25 commented May 12, 2018 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

maropu commented May 14, 2018

kiszk commented May 14, 2018

kiszk May 14, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented May 14, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cxzl25 commented May 14, 2018

SparkQA commented May 14, 2018

SparkQA commented May 14, 2018

JoshRosen commented May 22, 2018

cxzl25 commented May 22, 2018

cloud-fan commented May 22, 2018

cxzl25 commented May 22, 2018

cloud-fan May 22, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan May 22, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented May 22, 2018

SparkQA commented May 22, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented May 22, 2018

SparkQA commented May 22, 2018

SparkQA commented May 22, 2018

maropu commented May 23, 2018

SparkQA commented May 23, 2018

kiszk commented May 23, 2018

SparkQA commented May 23, 2018

kiszk commented May 23, 2018

SparkQA commented May 23, 2018

vanzin commented May 23, 2018

SparkQA commented May 23, 2018

gatorsmile commented May 23, 2018

SparkQA commented May 24, 2018

cloud-fan commented May 24, 2018 • edited Loading

cxzl25 commented May 24, 2018

cxzl25 commented May 12, 2018 •

edited

Loading

kiszk May 14, 2018 •

edited

Loading

cloud-fan May 22, 2018 •

edited

Loading

cloud-fan May 22, 2018 •

edited

Loading

cloud-fan commented May 24, 2018 •

edited

Loading