Skip to content

[SPARK-16215][SQL] Reduce runtime overhead of a program that writes an primitive array in Dataframe/Dataset #13911

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
kiszk wants to merge 10 commits into from
Closed

[SPARK-16215][SQL] Reduce runtime overhead of a program that writes an primitive array in Dataframe/Dataset #13911

Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
1c57972
pass information on containsNull in ArrayType
kiszk Jun 26, 2016
3d78991
generate helper call instead of for-loop
kiszk Jun 26, 2016
90d716f
add test suite
kiszk Jun 26, 2016
0b50242
fix test failure
kiszk Jun 26, 2016
4b132de
update
kiszk Jul 2, 2016
249a1a9
add test suites
kiszk Jul 2, 2016
96b0853
fix potential bug since writePrimitiveTypeArray() may receive Generic…
kiszk Jul 14, 2016
88aad46
rebase
kiszk Nov 3, 2016
25ddd8e
fix test failures
kiszk Nov 3, 2016
6204298
fix test failures
kiszk Nov 4, 2016
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
fix potential bug since writePrimitiveTypeArray() may receive Generic… 
…RefArrayData
  • Loading branch information
@kiszk
kiszk committed Nov 3, 2016
commit 96b0853d2876ba3aa3cc1db1c23df55e226deb49
135 changes: 117 additions & 18 deletions ...st/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UnsafeArrayWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,11 +262,7 @@ public void write(int ordinal, CalendarInterval input) {
}

private void writePrimitiveArray(Object input, int offset, int elementSize, int length) {
// uncomment this if SPARK-16043 is merged
// Platform.copyMemory(input, offset, holder.buffer, startingOffset + headerInBytes,
// elementSize * length);
Platform.copyMemory(input, offset, holder.buffer, holder.cursor, elementSize * length);
// remove the followings if SPARK-15962 is merged
for (int ordinal = 0; ordinal < length; ordinal++) {
Platform.putInt(holder.buffer, getElementOffset(ordinal),
(holder.cursor + ordinal * elementSize) - startingOffset);
Expand All @@ -275,65 +271,168 @@ private void writePrimitiveArray(Object input, int offset, int elementSize, int
}

public void writePrimitiveBooleanArray(ArrayData arrayData) {
// uncomment this if SPARK-16043 is merged
// boolean[] intput = ((GenericBooleanArrayData)arrayData).primitiveArray();
boolean[] input = arrayData.toBooleanArray();
int length = input.length;
int offset = Platform.BYTE_ARRAY_OFFSET;
writePrimitiveArray(input, offset, 1, length);
}

public void writePrimitiveByteArray(ArrayData arrayData) {
// uncomment this if SPARK-16043 is merged
// byte[] intput = ((GenericByteArrayData)arrayData).primitiveArray();
byte[] input = arrayData.toByteArray();
int length = input.length;
int offset = Platform.BYTE_ARRAY_OFFSET;
writePrimitiveArray(input, offset, 1, length);
}

public void writePrimitiveShortArray(ArrayData arrayData) {
// uncomment this if SPARK-16043 is merged
// short[] input = ((GenericShortArrayData)arrayData).primitiveArray();
short[] input = arrayData.toShortArray();
int length = input.length;
int offset = Platform.SHORT_ARRAY_OFFSET;
writePrimitiveArray(input, offset, 2, length);
}

public void writePrimitiveIntArray(ArrayData arrayData) {
// uncomment this if SPARK-16043 is merged
// int[] input = ((GenericIntArrayData)arrayData).primitiveArray();
int[] input = arrayData.toIntArray();
int length = input.length;
int offset = Platform.INT_ARRAY_OFFSET;
writePrimitiveArray(input, offset, 4, length);
}

public void writePrimitiveLongArray(ArrayData arrayData) {
// uncomment this if SPARK-16043 is merged
// long[] input = ((GenericLongArrayData)arrayData).primitiveArray();
long[] input = arrayData.toLongArray();
int length = input.length;
int offset = Platform.LONG_ARRAY_OFFSET;
writePrimitiveArray(input, offset, 8, length);
}

public void writePrimitiveFloatArray(ArrayData arrayData) {
// uncomment this if SPARK-16043 is merged
// float[] input = ((GenericFloatArrayData)arrayData).primitiveArray();
float[] input = arrayData.toFloatArray();
int length = input.length;
int offset = Platform.FLOAT_ARRAY_OFFSET;
writePrimitiveArray(input, offset, 4, length);
}

public void writePrimitiveDoubleArray(ArrayData arrayData) {
// uncomment this if SPARK-16043 is merged
// double[] input = ((GenericDoubleArrayData)arrayData).primitiveArray();
double[] input = arrayData.toDoubleArray();
int length = input.length;
int offset = Platform.DOUBLE_ARRAY_OFFSET;
writePrimitiveArray(input, offset, 8, length);
}

/** uncomment this if SPARK-16043 is merged

// remove this method if SPARK-15962 is merged
private void updateIndex(int elementSize, int length) {
for (int ordinal = 0; ordinal < length; ordinal++) {
Platform.putInt(holder.buffer, getElementOffset(ordinal),
(holder.cursor + ordinal * elementSize) - startingOffset);
}
holder.cursor += elementSize * length;
}

public void writePrimitiveBooleanArray(ArrayData arrayData) {
if (arrayData instanceof GenericBooleanArrayData) {
boolean[] input = ((GenericBooleanArrayData)arrayData).primitiveArray();
int length = input.length;
Platform.copyMemory(input, Platform.BOOLEAN_ARRAY_OFFSET,
holder.buffer, startingOffset + headerInBytes, length);
} else {
int length = arrayData.numElements();
for (int i = 0; i < length; i++) {
Platform.putBoolean(holder.buffer, holder.cursor + i, arrayData.getBoolean(i));
}
updateIndex(1, length);
}
}

public void writePrimitiveByteArray(ArrayData arrayData) {
if (arrayData instanceof GenericByteArrayData) {
byte[] input = ((GenericByteArrayData)arrayData).primitiveArray();
int length = input.length;
Platform.copyMemory(input, Platform.BYTE_ARRAY_OFFSET,
holder.buffer, startingOffset + headerInBytes, length);
} else {
int length = arrayData.numElements();
for (int i = 0; i < length; i++) {
Platform.putByte(holder.buffer, holder.cursor + i, arrayData.getByte(i));
}
updateIndex(1, length);
}
}

public void writePrimitiveShortArray(ArrayData arrayData) {
if (arrayData instanceof GenericShortArrayData) {
short[] input = ((GenericShortArrayData)arrayData).primitiveArray();
int length = input.length;
Platform.copyMemory(input, Platform.SHORT_ARRAY_OFFSET,
holder.buffer, startingOffset + headerInBytes, length);
} else {
int length = arrayData.numElements();
for (int i = 0; i < length; i++) {
Platform.putShort(holder.buffer, holder.cursor + i, arrayData.getShort(i));
}
updateIndex(2, length);
}
}

public void writePrimitiveIntArray(ArrayData arrayData) {
if (arrayData instanceof GenericIntArrayData) {
int[] input = ((GenericIntArrayData)arrayData).primitiveArray();
int length = input.length;
Platform.copyMemory(input, Platform.INT_ARRAY_OFFSET,
holder.buffer, startingOffset + headerInBytes, length);
} else {
int length = arrayData.numElements();
for (int i = 0; i < length; i++) {
Platform.putInt(holder.buffer, holder.cursor + i, arrayData.getInt(i));
}
updateIndex(4, length);
}
}

public void writePrimitiveLongArray(ArrayData arrayData) {
if (arrayData instanceof GenericLongArrayData) {
long[] input = ((GenericLongArrayData)arrayData).primitiveArray();
int length = input.length;
Platform.copyMemory(input, Platform.LONG_ARRAY_OFFSET,
holder.buffer, startingOffset + headerInBytes, length);
} else {
int length = arrayData.numElements();
for (int i = 0; i < length; i++) {
Platform.putLong(holder.buffer, holder.cursor + i, arrayData.getLong(i));
}
updateIndex(8, length);
}
}

public void writePrimitiveFloatArray(ArrayData arrayData) {
if (arrayData instanceof GenericFloatArrayData) {
float[] input = ((GenericFloatArrayData)arrayData).primitiveArray();
int length = input.length;
Platform.copyMemory(input, Platform.FLOAT_ARRAY_OFFSET,
holder.buffer, startingOffset + headerInBytes, length);
} else {
int length = arrayData.numElements();
for (int i = 0; i < length; i++) {
Platform.putFloat(holder.buffer, holder.cursor + i, arrayData.getFloat(i));
}
updateIndex(4, length);
}
}

public void writePrimitiveDoubleArray(ArrayData arrayData) {
if (arrayData instanceof GenericDoubleArrayData) {
double[] input = ((GenericDoubleArrayData)arrayData).primitiveArray();
int length = input.length;
Platform.copyMemory(input, Platform.DOUBLE_ARRAY_OFFSET,
holder.buffer, startingOffset + headerInBytes, length);
} else {
int length = arrayData.numElements();
for (int i = 0; i < length; i++) {
Platform.putFloat(holder.buffer, holder.cursor + i, arrayData.getFloat(i));
}
updateIndex(8, length);
}
}
*/
}