Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
59 commits
Select commit Hold shift + click to select a range
480a74a
Initial import of code from Databricks unsafe utils repo.
JoshRosen Apr 17, 2015
ab68e08
Begin merging the UTF8String implementations.
JoshRosen Apr 18, 2015
f03e9c1
Play around with Unsafe implementations of more string methods.
JoshRosen Apr 18, 2015
5d55cef
Add skeleton for Row implementation.
JoshRosen Apr 18, 2015
8a8f9df
Add skeleton for GeneratedAggregate integration.
JoshRosen Apr 18, 2015
1ff814d
Add reminder to free memory on iterator completion
JoshRosen Apr 18, 2015
53ba9b7
Start prototyping Java Row -> UnsafeRow converters
JoshRosen Apr 19, 2015
fc4c3a8
Sketch how the converters will be used in UnsafeGeneratedAggregate
JoshRosen Apr 19, 2015
1a483c5
First version that passes some aggregation tests:
JoshRosen Apr 19, 2015
079f1bf
Some clarification of the BytesToBytesMap.lookup() / set() contract.
JoshRosen Apr 19, 2015
f764d13
Simplify address + length calculation in Location.
JoshRosen Apr 19, 2015
c754ae1
Now that the store*() contract has been stregthened, we can remove an…
JoshRosen Apr 19, 2015
ae39694
Add finalizer as "cleanup method of last resort"
JoshRosen Apr 19, 2015
c7f0b56
Reuse UnsafeRow pointer in UnsafeRowConverter
JoshRosen Apr 20, 2015
62ab054
Optimize for fact that get() is only called on String columns.
JoshRosen Apr 20, 2015
c55bf66
Free buffer once iterator has been fully consumed.
JoshRosen Apr 20, 2015
738fa33
Add feature flag to guard UnsafeGeneratedAggregate
JoshRosen Apr 20, 2015
c1b3813
Fix bug in UnsafeMemoryAllocator.free():
JoshRosen Apr 20, 2015
7df6008
Optimizations related to zeroing out memory:
JoshRosen Apr 21, 2015
58ac393
Use UNSAFE allocator in GeneratedAggregate (TODO: make this configura…
JoshRosen Apr 21, 2015
d2bb986
Update to implement new Row methods added upstream
JoshRosen Apr 22, 2015
b3eaccd
Extract aggregation map into its own class.
JoshRosen Apr 22, 2015
bade966
Comment update (bumping to refresh GitHub cache...)
JoshRosen Apr 22, 2015
d85eeff
Add basic sanity test for UnsafeFixedWidthAggregationMap
JoshRosen Apr 22, 2015
1f4b716
Merge Unsafe code into the regular GeneratedAggregate, guarded by a
JoshRosen Apr 22, 2015
92d5a06
Address a number of minor code review comments.
JoshRosen Apr 23, 2015
628f936
Use ints intead of longs for indexing.
JoshRosen Apr 23, 2015
23a440a
Bump up default hash map size
JoshRosen Apr 23, 2015
765243d
Enable optional performance metrics for hash map.
JoshRosen Apr 23, 2015
b26f1d3
Fix bug in murmur hash implementation.
JoshRosen Apr 23, 2015
49aed30
More long -> int conversion.
JoshRosen Apr 23, 2015
29a7575
Remove debug logging
JoshRosen Apr 24, 2015
ef6b3d3
Fix a bunch of FindBugs and IntelliJ inspections
JoshRosen Apr 24, 2015
06e929d
More warning cleanup
JoshRosen Apr 24, 2015
854201a
Import and comment cleanup
JoshRosen Apr 24, 2015
f3dcbfe
More mod replacement
JoshRosen Apr 24, 2015
afe8dca
Some Javadoc cleanup
JoshRosen Apr 24, 2015
a95291e
Cleanups to string handling code
JoshRosen Apr 24, 2015
31eaabc
Lots of TODO and doc cleanup.
JoshRosen Apr 24, 2015
6ffdaa1
Null handling improvements in UnsafeRow.
JoshRosen Apr 24, 2015
9c19fc0
Add configuration options for heap vs. offheap
JoshRosen Apr 24, 2015
cde4132
Add missing pom.xml
JoshRosen Apr 26, 2015
0925847
Disable MiMa checks for new unsafe module
JoshRosen Apr 27, 2015
a8e4a3f
Introduce MemoryManager interface; add to SparkEnv.
JoshRosen Apr 28, 2015
b45f070
Don't redundantly store the offset from key to value, since we can co…
JoshRosen Apr 28, 2015
162caf7
Fix test compilation
JoshRosen Apr 28, 2015
3ca84b2
Only zero the used portion of groupingKeyConversionScratchSpace
JoshRosen Apr 28, 2015
529e571
Measure timeSpentResizing in nanoseconds instead of milliseconds.
JoshRosen Apr 28, 2015
ce3c565
More comments, formatting, and code cleanup.
JoshRosen Apr 28, 2015
78a5b84
Add logging to MemoryManager
JoshRosen Apr 28, 2015
a19e066
Rename unsafe Java test suites to match Scala test naming convention.
JoshRosen Apr 28, 2015
de5e001
Fix debug vs. trace in logging message.
JoshRosen Apr 28, 2015
6e4b192
Remove an unused method from ByteArrayMethods.
JoshRosen Apr 28, 2015
70a39e4
Split MemoryManager into ExecutorMemoryManager and TaskMemoryManager:
JoshRosen Apr 28, 2015
50e9671
Throw memory leak warning even in case of error; add warning about co…
JoshRosen Apr 29, 2015
017b2dc
Remove BytesToBytesMap.finalize()
JoshRosen Apr 29, 2015
1bc36cc
Refactor UnsafeRowConverter to avoid unnecessary boxing.
JoshRosen Apr 29, 2015
81f34f8
Follow 'place children last' convention for GeneratedAggregate
JoshRosen Apr 29, 2015
eeee512
Add converters for Null, Boolean, Byte, and Short columns.
JoshRosen Apr 29, 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
Begin merging the UTF8String implementations.
  • Loading branch information
JoshRosen committed Apr 22, 2015
commit ab68e081eef12333ff6c475cd70759ab7c6aeb74
5 changes: 5 additions & 0 deletions core/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,11 @@
<artifactId>spark-network-shuffle_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-unsafe_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>net.java.dev.jets3t</groupId>
<artifactId>jets3t</artifactId>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,9 @@ package org.apache.spark.sql.types

import java.util.Arrays

import org.apache.spark.unsafe.PlatformDependent
import org.apache.spark.unsafe.string.{UTF8StringPointer, UTF8StringMethods}

/**
* A UTF-8 String, as internal representation of StringType in SparkSQL
*
Expand All @@ -32,45 +35,31 @@ final class UTF8String extends Ordered[UTF8String] with Serializable {

private[this] var bytes: Array[Byte] = _

private val pointer: UTF8StringPointer = new UTF8StringPointer

/**
* Update the UTF8String with String.
*/
def set(str: String): UTF8String = {
bytes = str.getBytes("utf-8")
this
set(str.getBytes("utf-8"))
}

/**
* Update the UTF8String with Array[Byte], which should be encoded in UTF-8
*/
def set(bytes: Array[Byte]): UTF8String = {
this.bytes = bytes
pointer.set(bytes, PlatformDependent.BYTE_ARRAY_OFFSET, bytes.length)
this
}

/**
* Return the number of bytes for a code point with the first byte as `b`
* @param b The first byte of a code point
*/
@inline
private[this] def numOfBytes(b: Byte): Int = {
val offset = (b & 0xFF) - 192
if (offset >= 0) UTF8String.bytesOfCodePointInUTF8(offset) else 1
}

/**
* Return the number of code points in it.
*
* This is only used by Substring() when `start` is negative.
*/
def length(): Int = {
var len = 0
var i: Int = 0
while (i < bytes.length) {
i += numOfBytes(bytes(i))
len += 1
}
len
pointer.getLengthInCodePoints
}

def getBytes: Array[Byte] = {
Expand All @@ -90,12 +79,12 @@ final class UTF8String extends Ordered[UTF8String] with Serializable {
var c = 0
var i: Int = 0
while (c < start && i < bytes.length) {
i += numOfBytes(bytes(i))
i += UTF8StringMethods.numOfBytes(bytes(i))
c += 1
}
var j = i
while (c < until && j < bytes.length) {
j += numOfBytes(bytes(j))
j += UTF8StringMethods.numOfBytes(bytes(j))
c += 1
}
UTF8String(Arrays.copyOfRange(bytes, i, j))
Expand Down Expand Up @@ -150,14 +139,14 @@ final class UTF8String extends Ordered[UTF8String] with Serializable {
override def clone(): UTF8String = new UTF8String().set(this.bytes)

override def compare(other: UTF8String): Int = {
var i: Int = 0
val b = other.getBytes
while (i < bytes.length && i < b.length) {
val res = bytes(i).compareTo(b(i))
if (res != 0) return res
i += 1
}
bytes.length - b.length
UTF8StringMethods.compare(
pointer.getBaseObject,
pointer.getBaseOffset,
pointer.getLengthInBytes,
other.pointer.getBaseObject,
other.pointer.getBaseOffset,
other.pointer.getLengthInBytes
)
}

override def compareTo(other: UTF8String): Int = {
Expand All @@ -181,14 +170,6 @@ final class UTF8String extends Ordered[UTF8String] with Serializable {
}

object UTF8String {
// number of tailing bytes in a UTF8 sequence for a code point
// see http://en.wikipedia.org/wiki/UTF-8, 192-256 of Byte 1
private[types] val bytesOfCodePointInUTF8: Array[Int] = Array(2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2,
2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2,
3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3,
4, 4, 4, 4, 4, 4, 4, 4,
5, 5, 5, 5,
6, 6, 6, 6)

/**
* Create a UTF-8 String from String
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,12 +40,44 @@ static long getLengthInBytes(Object baseObject, long baseOffset) {
return PlatformDependent.UNSAFE.getLong(baseObject, baseOffset);
}

public static String toJavaString(Object baseObject, long baseOffset) {
final long lengthInBytes = getLengthInBytes(baseObject, baseOffset);
public static int compare(
Object leftBaseObject,
long leftBaseOffset,
int leftBaseLengthInBytes,
Object rightBaseObject,
long rightBaseOffset,
int rightBaseLengthInBytes) {
int i = 0;
while (i < leftBaseLengthInBytes && i < rightBaseLengthInBytes) {
final byte leftByte = PlatformDependent.UNSAFE.getByte(leftBaseObject, leftBaseOffset + i);
final byte rightByte = PlatformDependent.UNSAFE.getByte(rightBaseObject, rightBaseOffset + i);
final int res = leftByte - rightByte;
if (res != 0) return res;
i += 1;
}
return leftBaseLengthInBytes - rightBaseLengthInBytes;
}

/**
* Return the number of code points in a string.
*
* This is only used by Substring() when `start` is negative.
*/
public static int getLengthInCodePoints(Object baseObject, long baseOffset, int lengthInBytes) {
int len = 0;
int i = 0;
while (i < lengthInBytes) {
i += numOfBytes(PlatformDependent.UNSAFE.getByte(baseObject, baseOffset + i));
len += 1;
}
return len;
}

public static String toJavaString(Object baseObject, long baseOffset, int lengthInBytes) {
final byte[] bytes = new byte[(int) lengthInBytes];
PlatformDependent.UNSAFE.copyMemory(
baseObject,
baseOffset + 8, // skip over the length
baseOffset,
bytes,
PlatformDependent.BYTE_ARRAY_OFFSET,
lengthInBytes
Expand All @@ -67,15 +99,40 @@ public static String toJavaString(Object baseObject, long baseOffset) {
public static long createFromJavaString(Object baseObject, long baseOffset, String str) {
final byte[] strBytes = str.getBytes();
final long strLengthInBytes = strBytes.length;
PlatformDependent.UNSAFE.putLong(baseObject, baseOffset, strLengthInBytes);
PlatformDependent.copyMemory(
strBytes,
PlatformDependent.BYTE_ARRAY_OFFSET,
baseObject,
baseOffset + 8,
baseOffset,
strLengthInBytes
);
return (8 + strLengthInBytes);
return strLengthInBytes;
}

/**
* Return the number of bytes for a code point with the first byte as `b`
* @param b The first byte of a code point
*/
public static int numOfBytes(byte b) {
final int offset = (b & 0xFF) - 192;
if (offset >= 0) {
return bytesOfCodePointInUTF8[offset];
} else {
return 1;
}
}

/**
* number of tailing bytes in a UTF8 sequence for a code point
* see http://en.wikipedia.org/wiki/UTF-8, 192-256 of Byte 1
*/
private static int[] bytesOfCodePointInUTF8 = new int[] {
2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2,
2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2,
3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3,
4, 4, 4, 4, 4, 4, 4, 4,
5, 5, 5, 5,
6, 6, 6, 6
};

}
Original file line number Diff line number Diff line change
Expand Up @@ -17,16 +17,39 @@

package org.apache.spark.unsafe.string;

import org.apache.spark.unsafe.memory.MemoryLocation;
import javax.annotation.Nullable;

/**
* A pointer to UTF8String data.
*/
public class UTF8StringPointer extends MemoryLocation {
public class UTF8StringPointer {

public long getLengthInBytes() { return UTF8StringMethods.getLengthInBytes(obj, offset); }
@Nullable
protected Object obj;
protected long offset;
protected int lengthInBytes;

public String toJavaString() { return UTF8StringMethods.toJavaString(obj, offset); }
public UTF8StringPointer() { }

public void set(Object obj, long offset, int lengthInBytes) {
this.obj = obj;
this.offset = offset;
this.lengthInBytes = lengthInBytes;
}

public int getLengthInCodePoints() {
return UTF8StringMethods.getLengthInCodePoints(obj, offset, lengthInBytes);
}

public int getLengthInBytes() { return lengthInBytes; }

public Object getBaseObject() { return obj; }

public long getBaseOffset() { return offset; }

public String toJavaString() {
return UTF8StringMethods.toJavaString(obj, offset, lengthInBytes);
}

@Override public String toString() { return toJavaString(); }
}
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,13 @@ public void toStringTest() {
final byte[] javaStrBytes = javaStr.getBytes();
final int paddedSizeInWords = javaStrBytes.length / 8 + (javaStrBytes.length % 8 == 0 ? 0 : 1);
final MemoryLocation memory = MemoryBlock.fromLongArray(new long[paddedSizeInWords]);
final long bytesWritten =
UTF8StringMethods.createFromJavaString(memory.getBaseObject(), memory.getBaseOffset(), javaStr);
Assert.assertEquals(8 + javaStrBytes.length, bytesWritten);
final long bytesWritten = UTF8StringMethods.createFromJavaString(
memory.getBaseObject(),
memory.getBaseOffset(),
javaStr);
Assert.assertEquals(javaStrBytes.length, bytesWritten);
final UTF8StringPointer utf8String = new UTF8StringPointer();
utf8String.setObjAndOffset(memory.getBaseObject(), memory.getBaseOffset());
utf8String.set(memory.getBaseObject(), memory.getBaseOffset(), bytesWritten);
Assert.assertEquals(javaStr, utf8String.toJavaString());
}
}