ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-10351] [SQL] Fixes UTF8String.fromAddress to handle off-heap memory

Browse source 
CC rxin marmbrus

Author: Feynman Liang <fliang@databricks.com>

Closes #8523 from feynmanliang/SPARK-10351.
This commit is contained in:
Feynman Liang 2015-08-30 23:12:56 -07:00 committed by Reynold Xin
parent 35e896a79b
commit 8694c3ad7d
2 changed files with 6 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
sql/core/src/test/scala/org/apache/spark/sql/UnsafeRowSuite.scala
View file

@ -43,12 +43,12 @@ class UnsafeRowSuite extends SparkFunSuite {
val arrayBackedUnsafeRow: UnsafeRow = val arrayBackedUnsafeRow: UnsafeRow =
UnsafeProjection.create(Array[DataType](StringType, StringType, IntegerType)).apply(row) UnsafeProjection.create(Array[DataType](StringType, StringType, IntegerType)).apply(row)
assert(arrayBackedUnsafeRow.getBaseObject.isInstanceOf[Array[Byte]]) assert(arrayBackedUnsafeRow.getBaseObject.isInstanceOf[Array[Byte]])
val bytesFromArrayBackedRow: Array[Byte] = { val (bytesFromArrayBackedRow, field0StringFromArrayBackedRow): (Array[Byte], String) = {
val baos = new ByteArrayOutputStream() val baos = new ByteArrayOutputStream()
arrayBackedUnsafeRow.writeToStream(baos, null) arrayBackedUnsafeRow.writeToStream(baos, null)
baos.toByteArray (baos.toByteArray, arrayBackedUnsafeRow.getString(0))
} }
val bytesFromOffheapRow: Array[Byte] = { val (bytesFromOffheapRow, field0StringFromOffheapRow): (Array[Byte], String) = {
val offheapRowPage = MemoryAllocator.UNSAFE.allocate(arrayBackedUnsafeRow.getSizeInBytes) val offheapRowPage = MemoryAllocator.UNSAFE.allocate(arrayBackedUnsafeRow.getSizeInBytes)
try { try {
Platform.copyMemory( Platform.copyMemory(
@ -69,13 +69,14 @@ class UnsafeRowSuite extends SparkFunSuite {
val baos = new ByteArrayOutputStream() val baos = new ByteArrayOutputStream()
val writeBuffer = new Array[Byte](1024) val writeBuffer = new Array[Byte](1024)
offheapUnsafeRow.writeToStream(baos, writeBuffer) offheapUnsafeRow.writeToStream(baos, writeBuffer)
baos.toByteArray (baos.toByteArray, offheapUnsafeRow.getString(0))
} finally { } finally {
MemoryAllocator.UNSAFE.free(offheapRowPage) MemoryAllocator.UNSAFE.free(offheapRowPage)
} }
} }
assert(bytesFromArrayBackedRow === bytesFromOffheapRow) assert(bytesFromArrayBackedRow === bytesFromOffheapRow)
assert(field0StringFromArrayBackedRow === field0StringFromOffheapRow)
} }
test("calling getDouble() and getFloat() on null columns") { test("calling getDouble() and getFloat() on null columns") {

6
unsafe/src/main/java/org/apache/spark/unsafe/types/UTF8String.java
View file

@ -90,11 +90,7 @@ public final class UTF8String implements Comparable<UTF8String>, Externalizable
* Creates an UTF8String from given address (base and offset) and length. * Creates an UTF8String from given address (base and offset) and length.
*/ */
public static UTF8String fromAddress(Object base, long offset, int numBytes) { public static UTF8String fromAddress(Object base, long offset, int numBytes) {
if (base != null) { return new UTF8String(base, offset, numBytes);
return new UTF8String(base, offset, numBytes);
} else {
return null;
}
} }
/** /**