You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2017/11/30 10:34:48 UTC
spark git commit: [SPARK-22643][SQL] ColumnarArray should be an
immutable view
Repository: spark
Updated Branches:
refs/heads/master 444a2bbb6 -> 9c29c5576
[SPARK-22643][SQL] ColumnarArray should be an immutable view
## What changes were proposed in this pull request?
To make `ColumnVector` public, `ColumnarArray` need to be public too, and we should not have mutable public fields in a public class. This PR proposes to make `ColumnarArray` an immutable view of the data, and always create a new instance of `ColumnarArray` in `ColumnVector#getArray`
## How was this patch tested?
new benchmark in `ColumnarBatchBenchmark`
Author: Wenchen Fan <we...@databricks.com>
Closes #19842 from cloud-fan/column-vector.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/9c29c557
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/9c29c557
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/9c29c557
Branch: refs/heads/master
Commit: 9c29c557635caf739fde942f53255273aac0d7b1
Parents: 444a2bb
Author: Wenchen Fan <we...@databricks.com>
Authored: Thu Nov 30 18:34:38 2017 +0800
Committer: Wenchen Fan <we...@databricks.com>
Committed: Thu Nov 30 18:34:38 2017 +0800
----------------------------------------------------------------------
.../parquet/VectorizedColumnReader.java | 2 +-
.../execution/vectorized/ArrowColumnVector.java | 1 -
.../sql/execution/vectorized/ColumnVector.java | 14 +-
.../execution/vectorized/ColumnVectorUtils.java | 18 +--
.../sql/execution/vectorized/ColumnarArray.java | 10 +-
.../vectorized/OffHeapColumnVector.java | 2 +-
.../vectorized/OnHeapColumnVector.java | 2 +-
.../vectorized/WritableColumnVector.java | 13 +-
.../vectorized/ColumnVectorSuite.scala | 41 +++---
.../vectorized/ColumnarBatchBenchmark.scala | 142 +++++++++++++++----
.../vectorized/ColumnarBatchSuite.scala | 18 +--
11 files changed, 164 insertions(+), 99 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/9c29c557/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
----------------------------------------------------------------------
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
index 0f1f470..71ca8b1 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
@@ -425,7 +425,7 @@ public class VectorizedColumnReader {
// This is where we implement support for the valid type conversions.
// TODO: implement remaining type conversions
VectorizedValuesReader data = (VectorizedValuesReader) dataColumn;
- if (column.isArray()) {
+ if (column.dataType() == DataTypes.StringType || column.dataType() == DataTypes.BinaryType) {
defColumn.readBinarys(num, column, rowId, maxDefLevel, data);
} else if (column.dataType() == DataTypes.TimestampType) {
for (int i = 0; i < num; i++) {
http://git-wip-us.apache.org/repos/asf/spark/blob/9c29c557/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ArrowColumnVector.java
----------------------------------------------------------------------
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ArrowColumnVector.java b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ArrowColumnVector.java
index 5c502c9..0071bd6 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ArrowColumnVector.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ArrowColumnVector.java
@@ -315,7 +315,6 @@ public final class ArrowColumnVector extends ColumnVector {
childColumns = new ArrowColumnVector[1];
childColumns[0] = new ArrowColumnVector(listVector.getDataVector());
- resultArray = new ColumnarArray(childColumns[0]);
} else if (vector instanceof MapVector) {
MapVector mapVector = (MapVector) vector;
accessor = new StructAccessor(mapVector);
http://git-wip-us.apache.org/repos/asf/spark/blob/9c29c557/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java
----------------------------------------------------------------------
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java
index 940457f..cca1491 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java
@@ -175,9 +175,7 @@ public abstract class ColumnVector implements AutoCloseable {
* Returns the array at rowid.
*/
public final ColumnarArray getArray(int rowId) {
- resultArray.length = getArrayLength(rowId);
- resultArray.offset = getArrayOffset(rowId);
- return resultArray;
+ return new ColumnarArray(arrayData(), getArrayOffset(rowId), getArrayLength(rowId));
}
/**
@@ -214,21 +212,11 @@ public abstract class ColumnVector implements AutoCloseable {
public abstract ColumnVector getChildColumn(int ordinal);
/**
- * Returns true if this column is an array.
- */
- public final boolean isArray() { return resultArray != null; }
-
- /**
* Data type for this column.
*/
protected DataType type;
/**
- * Reusable Array holder for getArray().
- */
- protected ColumnarArray resultArray;
-
- /**
* Reusable Struct holder for getStruct().
*/
protected ColumnarRow resultStruct;
http://git-wip-us.apache.org/repos/asf/spark/blob/9c29c557/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVectorUtils.java
----------------------------------------------------------------------
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVectorUtils.java b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVectorUtils.java
index b4b5f0a..bc62bc4 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVectorUtils.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVectorUtils.java
@@ -98,21 +98,13 @@ public class ColumnVectorUtils {
* For example, an array of IntegerType will return an int[].
* Throws exceptions for unhandled schemas.
*/
- public static Object toPrimitiveJavaArray(ColumnarArray array) {
- DataType dt = array.data.dataType();
- if (dt instanceof IntegerType) {
- int[] result = new int[array.length];
- ColumnVector data = array.data;
- for (int i = 0; i < result.length; i++) {
- if (data.isNullAt(array.offset + i)) {
- throw new RuntimeException("Cannot handle NULL values.");
- }
- result[i] = data.getInt(array.offset + i);
+ public static int[] toJavaIntArray(ColumnarArray array) {
+ for (int i = 0; i < array.numElements(); i++) {
+ if (array.isNullAt(i)) {
+ throw new RuntimeException("Cannot handle NULL values.");
}
- return result;
- } else {
- throw new UnsupportedOperationException();
}
+ return array.toIntArray();
}
private static void appendValue(WritableColumnVector dst, DataType t, Object o) {
http://git-wip-us.apache.org/repos/asf/spark/blob/9c29c557/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarArray.java
----------------------------------------------------------------------
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarArray.java b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarArray.java
index b9da641..cbc39d1 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarArray.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarArray.java
@@ -29,12 +29,14 @@ import org.apache.spark.unsafe.types.UTF8String;
public final class ColumnarArray extends ArrayData {
// The data for this array. This array contains elements from
// data[offset] to data[offset + length).
- public final ColumnVector data;
- public int length;
- public int offset;
+ private final ColumnVector data;
+ private final int offset;
+ private final int length;
- ColumnarArray(ColumnVector data) {
+ ColumnarArray(ColumnVector data, int offset, int length) {
this.data = data;
+ this.offset = offset;
+ this.length = length;
}
@Override
http://git-wip-us.apache.org/repos/asf/spark/blob/9c29c557/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java
----------------------------------------------------------------------
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java
index 1cbaf08..806d029 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java
@@ -532,7 +532,7 @@ public final class OffHeapColumnVector extends WritableColumnVector {
@Override
protected void reserveInternal(int newCapacity) {
int oldCapacity = (nulls == 0L) ? 0 : capacity;
- if (this.resultArray != null) {
+ if (isArray()) {
this.lengthData =
Platform.reallocateMemory(lengthData, oldCapacity * 4, newCapacity * 4);
this.offsetData =
http://git-wip-us.apache.org/repos/asf/spark/blob/9c29c557/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java
----------------------------------------------------------------------
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java
index 85d7229..6e7f74c 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java
@@ -505,7 +505,7 @@ public final class OnHeapColumnVector extends WritableColumnVector {
// Spilt this function out since it is the slow path.
@Override
protected void reserveInternal(int newCapacity) {
- if (this.resultArray != null || DecimalType.isByteArrayDecimalType(type)) {
+ if (isArray()) {
int[] newLengths = new int[newCapacity];
int[] newOffsets = new int[newCapacity];
if (this.arrayLengths != null) {
http://git-wip-us.apache.org/repos/asf/spark/blob/9c29c557/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java
----------------------------------------------------------------------
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java
index e7653f0..0bea4cc 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java
@@ -75,7 +75,6 @@ public abstract class WritableColumnVector extends ColumnVector {
}
dictionary = null;
resultStruct = null;
- resultArray = null;
}
public void reserve(int requiredCapacity) {
@@ -650,6 +649,11 @@ public abstract class WritableColumnVector extends ColumnVector {
*/
protected abstract WritableColumnVector reserveNewColumn(int capacity, DataType type);
+ protected boolean isArray() {
+ return type instanceof ArrayType || type instanceof BinaryType || type instanceof StringType ||
+ DecimalType.isByteArrayDecimalType(type);
+ }
+
/**
* Sets up the common state and also handles creating the child columns if this is a nested
* type.
@@ -658,8 +662,7 @@ public abstract class WritableColumnVector extends ColumnVector {
super(type);
this.capacity = capacity;
- if (type instanceof ArrayType || type instanceof BinaryType || type instanceof StringType
- || DecimalType.isByteArrayDecimalType(type)) {
+ if (isArray()) {
DataType childType;
int childCapacity = capacity;
if (type instanceof ArrayType) {
@@ -670,7 +673,6 @@ public abstract class WritableColumnVector extends ColumnVector {
}
this.childColumns = new WritableColumnVector[1];
this.childColumns[0] = reserveNewColumn(childCapacity, childType);
- this.resultArray = new ColumnarArray(this.childColumns[0]);
this.resultStruct = null;
} else if (type instanceof StructType) {
StructType st = (StructType)type;
@@ -678,18 +680,15 @@ public abstract class WritableColumnVector extends ColumnVector {
for (int i = 0; i < childColumns.length; ++i) {
this.childColumns[i] = reserveNewColumn(capacity, st.fields()[i].dataType());
}
- this.resultArray = null;
this.resultStruct = new ColumnarRow(this.childColumns);
} else if (type instanceof CalendarIntervalType) {
// Two columns. Months as int. Microseconds as Long.
this.childColumns = new WritableColumnVector[2];
this.childColumns[0] = reserveNewColumn(capacity, DataTypes.IntegerType);
this.childColumns[1] = reserveNewColumn(capacity, DataTypes.LongType);
- this.resultArray = null;
this.resultStruct = new ColumnarRow(this.childColumns);
} else {
this.childColumns = null;
- this.resultArray = null;
this.resultStruct = null;
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/9c29c557/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnVectorSuite.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnVectorSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnVectorSuite.scala
index e28ab71..54b31ce 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnVectorSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnVectorSuite.scala
@@ -57,7 +57,7 @@ class ColumnVectorSuite extends SparkFunSuite with BeforeAndAfterEach {
testVector.appendBoolean(i % 2 == 0)
}
- val array = new ColumnarArray(testVector)
+ val array = new ColumnarArray(testVector, 0, 10)
(0 until 10).foreach { i =>
assert(array.get(i, BooleanType) === (i % 2 == 0))
@@ -69,7 +69,7 @@ class ColumnVectorSuite extends SparkFunSuite with BeforeAndAfterEach {
testVector.appendByte(i.toByte)
}
- val array = new ColumnarArray(testVector)
+ val array = new ColumnarArray(testVector, 0, 10)
(0 until 10).foreach { i =>
assert(array.get(i, ByteType) === i.toByte)
@@ -81,7 +81,7 @@ class ColumnVectorSuite extends SparkFunSuite with BeforeAndAfterEach {
testVector.appendShort(i.toShort)
}
- val array = new ColumnarArray(testVector)
+ val array = new ColumnarArray(testVector, 0, 10)
(0 until 10).foreach { i =>
assert(array.get(i, ShortType) === i.toShort)
@@ -93,7 +93,7 @@ class ColumnVectorSuite extends SparkFunSuite with BeforeAndAfterEach {
testVector.appendInt(i)
}
- val array = new ColumnarArray(testVector)
+ val array = new ColumnarArray(testVector, 0, 10)
(0 until 10).foreach { i =>
assert(array.get(i, IntegerType) === i)
@@ -105,7 +105,7 @@ class ColumnVectorSuite extends SparkFunSuite with BeforeAndAfterEach {
testVector.appendLong(i)
}
- val array = new ColumnarArray(testVector)
+ val array = new ColumnarArray(testVector, 0, 10)
(0 until 10).foreach { i =>
assert(array.get(i, LongType) === i)
@@ -117,7 +117,7 @@ class ColumnVectorSuite extends SparkFunSuite with BeforeAndAfterEach {
testVector.appendFloat(i.toFloat)
}
- val array = new ColumnarArray(testVector)
+ val array = new ColumnarArray(testVector, 0, 10)
(0 until 10).foreach { i =>
assert(array.get(i, FloatType) === i.toFloat)
@@ -129,7 +129,7 @@ class ColumnVectorSuite extends SparkFunSuite with BeforeAndAfterEach {
testVector.appendDouble(i.toDouble)
}
- val array = new ColumnarArray(testVector)
+ val array = new ColumnarArray(testVector, 0, 10)
(0 until 10).foreach { i =>
assert(array.get(i, DoubleType) === i.toDouble)
@@ -142,7 +142,7 @@ class ColumnVectorSuite extends SparkFunSuite with BeforeAndAfterEach {
testVector.appendByteArray(utf8, 0, utf8.length)
}
- val array = new ColumnarArray(testVector)
+ val array = new ColumnarArray(testVector, 0, 10)
(0 until 10).foreach { i =>
assert(array.get(i, StringType) === UTF8String.fromString(s"str$i"))
@@ -155,7 +155,7 @@ class ColumnVectorSuite extends SparkFunSuite with BeforeAndAfterEach {
testVector.appendByteArray(utf8, 0, utf8.length)
}
- val array = new ColumnarArray(testVector)
+ val array = new ColumnarArray(testVector, 0, 10)
(0 until 10).foreach { i =>
val utf8 = s"str$i".getBytes("utf8")
@@ -191,12 +191,10 @@ class ColumnVectorSuite extends SparkFunSuite with BeforeAndAfterEach {
testVector.putArray(2, 3, 0)
testVector.putArray(3, 3, 3)
- val array = new ColumnarArray(testVector)
-
- assert(array.getArray(0).toIntArray() === Array(0))
- assert(array.getArray(1).toIntArray() === Array(1, 2))
- assert(array.getArray(2).toIntArray() === Array.empty[Int])
- assert(array.getArray(3).toIntArray() === Array(3, 4, 5))
+ assert(testVector.getArray(0).toIntArray() === Array(0))
+ assert(testVector.getArray(1).toIntArray() === Array(1, 2))
+ assert(testVector.getArray(2).toIntArray() === Array.empty[Int])
+ assert(testVector.getArray(3).toIntArray() === Array(3, 4, 5))
}
val structType: StructType = new StructType().add("int", IntegerType).add("double", DoubleType)
@@ -208,12 +206,10 @@ class ColumnVectorSuite extends SparkFunSuite with BeforeAndAfterEach {
c1.putInt(1, 456)
c2.putDouble(1, 5.67)
- val array = new ColumnarArray(testVector)
-
- assert(array.getStruct(0, structType.length).get(0, IntegerType) === 123)
- assert(array.getStruct(0, structType.length).get(1, DoubleType) === 3.45)
- assert(array.getStruct(1, structType.length).get(0, IntegerType) === 456)
- assert(array.getStruct(1, structType.length).get(1, DoubleType) === 5.67)
+ assert(testVector.getStruct(0, structType.length).get(0, IntegerType) === 123)
+ assert(testVector.getStruct(0, structType.length).get(1, DoubleType) === 3.45)
+ assert(testVector.getStruct(1, structType.length).get(0, IntegerType) === 456)
+ assert(testVector.getStruct(1, structType.length).get(1, DoubleType) === 5.67)
}
test("[SPARK-22092] off-heap column vector reallocation corrupts array data") {
@@ -226,9 +222,8 @@ class ColumnVectorSuite extends SparkFunSuite with BeforeAndAfterEach {
testVector.reserve(16)
// Check that none of the values got lost/overwritten.
- val array = new ColumnarArray(testVector)
(0 until 8).foreach { i =>
- assert(array.get(i, arrayType).asInstanceOf[ArrayData].toIntArray() === Array(i))
+ assert(testVector.getArray(i).toIntArray() === Array(i))
}
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/9c29c557/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchBenchmark.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchBenchmark.scala
index 705b26b..38ea2e4 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchBenchmark.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchBenchmark.scala
@@ -23,11 +23,9 @@ import scala.util.Random
import org.apache.spark.memory.MemoryMode
import org.apache.spark.sql.catalyst.expressions.UnsafeRow
-import org.apache.spark.sql.execution.vectorized.ColumnVector
import org.apache.spark.sql.execution.vectorized.OffHeapColumnVector
import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector
-import org.apache.spark.sql.execution.vectorized.WritableColumnVector
-import org.apache.spark.sql.types.{BinaryType, DataType, IntegerType}
+import org.apache.spark.sql.types.{ArrayType, BinaryType, IntegerType}
import org.apache.spark.unsafe.Platform
import org.apache.spark.util.Benchmark
import org.apache.spark.util.collection.BitSet
@@ -265,20 +263,22 @@ object ColumnarBatchBenchmark {
}
/*
- Intel(R) Core(TM) i7-4870HQ CPU @ 2.50GHz
- Int Read/Write: Avg Time(ms) Avg Rate(M/s) Relative Rate
- -------------------------------------------------------------------------
- Java Array 248.8 1317.04 1.00 X
- ByteBuffer Unsafe 435.6 752.25 0.57 X
- ByteBuffer API 1752.0 187.03 0.14 X
- DirectByteBuffer 595.4 550.35 0.42 X
- Unsafe Buffer 235.2 1393.20 1.06 X
- Column(on heap) 189.8 1726.45 1.31 X
- Column(off heap) 408.4 802.35 0.61 X
- Column(off heap direct) 237.6 1379.12 1.05 X
- UnsafeRow (on heap) 414.6 790.35 0.60 X
- UnsafeRow (off heap) 487.2 672.58 0.51 X
- Column On Heap Append 530.1 618.14 0.59 X
+ Java HotSpot(TM) 64-Bit Server VM 1.8.0_60-b27 on Mac OS X 10.13.1
+ Intel(R) Core(TM) i7-4960HQ CPU @ 2.60GHz
+
+ Int Read/Write: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative
+ ------------------------------------------------------------------------------------------------
+ Java Array 177 / 181 1856.4 0.5 1.0X
+ ByteBuffer Unsafe 318 / 322 1032.0 1.0 0.6X
+ ByteBuffer API 1411 / 1418 232.2 4.3 0.1X
+ DirectByteBuffer 467 / 474 701.8 1.4 0.4X
+ Unsafe Buffer 178 / 185 1843.6 0.5 1.0X
+ Column(on heap) 178 / 184 1840.8 0.5 1.0X
+ Column(off heap) 341 / 344 961.8 1.0 0.5X
+ Column(off heap direct) 178 / 184 1845.4 0.5 1.0X
+ UnsafeRow (on heap) 378 / 389 866.3 1.2 0.5X
+ UnsafeRow (off heap) 393 / 402 834.0 1.2 0.4X
+ Column On Heap Append 309 / 318 1059.1 0.9 0.6X
*/
val benchmark = new Benchmark("Int Read/Write", count * iters)
benchmark.addCase("Java Array")(javaArray)
@@ -332,11 +332,13 @@ object ColumnarBatchBenchmark {
}
}}
/*
- Intel(R) Core(TM) i7-4870HQ CPU @ 2.50GHz
- Boolean Read/Write: Avg Time(ms) Avg Rate(M/s) Relative Rate
- -------------------------------------------------------------------------
- Bitset 895.88 374.54 1.00 X
- Byte Array 578.96 579.56 1.55 X
+ Java HotSpot(TM) 64-Bit Server VM 1.8.0_60-b27 on Mac OS X 10.13.1
+ Intel(R) Core(TM) i7-4960HQ CPU @ 2.60GHz
+
+ Boolean Read/Write: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative
+ ------------------------------------------------------------------------------------------------
+ Bitset 726 / 727 462.4 2.2 1.0X
+ Byte Array 530 / 542 632.7 1.6 1.4X
*/
benchmark.run()
}
@@ -387,10 +389,13 @@ object ColumnarBatchBenchmark {
}
/*
- String Read/Write: Avg Time(ms) Avg Rate(M/s) Relative Rate
- -------------------------------------------------------------------------------------
- On Heap 457.0 35.85 1.00 X
- Off Heap 1206.0 13.59 0.38 X
+ Java HotSpot(TM) 64-Bit Server VM 1.8.0_60-b27 on Mac OS X 10.13.1
+ Intel(R) Core(TM) i7-4960HQ CPU @ 2.60GHz
+
+ String Read/Write: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative
+ ------------------------------------------------------------------------------------------------
+ On Heap 332 / 338 49.3 20.3 1.0X
+ Off Heap 466 / 467 35.2 28.4 0.7X
*/
val benchmark = new Benchmark("String Read/Write", count * iters)
benchmark.addCase("On Heap")(column(MemoryMode.ON_HEAP))
@@ -398,9 +403,94 @@ object ColumnarBatchBenchmark {
benchmark.run
}
+ def arrayAccess(iters: Int): Unit = {
+ val random = new Random(0)
+ val count = 4 * 1000
+
+ val onHeapVector = new OnHeapColumnVector(count, ArrayType(IntegerType))
+ val offHeapVector = new OffHeapColumnVector(count, ArrayType(IntegerType))
+
+ val minSize = 3
+ val maxSize = 32
+ var arraysCount = 0
+ var elementsCount = 0
+ while (arraysCount < count) {
+ val size = random.nextInt(maxSize - minSize) + minSize
+ val onHeapArrayData = onHeapVector.arrayData()
+ val offHeapArrayData = offHeapVector.arrayData()
+
+ var i = 0
+ while (i < size) {
+ val value = random.nextInt()
+ onHeapArrayData.appendInt(value)
+ offHeapArrayData.appendInt(value)
+ i += 1
+ }
+
+ onHeapVector.putArray(arraysCount, elementsCount, size)
+ offHeapVector.putArray(arraysCount, elementsCount, size)
+ elementsCount += size
+ arraysCount += 1
+ }
+
+ def readArrays(onHeap: Boolean): Unit = {
+ System.gc()
+ val vector = if (onHeap) onHeapVector else offHeapVector
+
+ var sum = 0L
+ for (_ <- 0 until iters) {
+ var i = 0
+ while (i < count) {
+ sum += vector.getArray(i).numElements()
+ i += 1
+ }
+ }
+ }
+
+ def readArrayElements(onHeap: Boolean): Unit = {
+ System.gc()
+ val vector = if (onHeap) onHeapVector else offHeapVector
+
+ var sum = 0L
+ for (_ <- 0 until iters) {
+ var i = 0
+ while (i < count) {
+ val array = vector.getArray(i)
+ val size = array.numElements()
+ var j = 0
+ while (j < size) {
+ sum += array.getInt(j)
+ j += 1
+ }
+ i += 1
+ }
+ }
+ }
+
+ val benchmark = new Benchmark("Array Vector Read", count * iters)
+ benchmark.addCase("On Heap Read Size Only") { _ => readArrays(true) }
+ benchmark.addCase("Off Heap Read Size Only") { _ => readArrays(false) }
+ benchmark.addCase("On Heap Read Elements") { _ => readArrayElements(true) }
+ benchmark.addCase("Off Heap Read Elements") { _ => readArrayElements(false) }
+
+ /*
+ Java HotSpot(TM) 64-Bit Server VM 1.8.0_60-b27 on Mac OS X 10.13.1
+ Intel(R) Core(TM) i7-4960HQ CPU @ 2.60GHz
+
+ Array Vector Read: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative
+ ------------------------------------------------------------------------------------------------
+ On Heap Read Size Only 415 / 422 394.7 2.5 1.0X
+ Off Heap Read Size Only 394 / 402 415.9 2.4 1.1X
+ On Heap Read Elements 2558 / 2593 64.0 15.6 0.2X
+ Off Heap Read Elements 3316 / 3317 49.4 20.2 0.1X
+ */
+ benchmark.run
+ }
+
def main(args: Array[String]): Unit = {
intAccess(1024 * 40)
booleanAccess(1024 * 40)
stringAccess(1024 * 4)
+ arrayAccess(1024 * 40)
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/9c29c557/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala
index 1b4e2ba..0ae4f2d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala
@@ -645,26 +645,26 @@ class ColumnarBatchSuite extends SparkFunSuite {
column.putArray(2, 2, 0)
column.putArray(3, 3, 3)
- val a1 = ColumnVectorUtils.toPrimitiveJavaArray(column.getArray(0)).asInstanceOf[Array[Int]]
- val a2 = ColumnVectorUtils.toPrimitiveJavaArray(column.getArray(1)).asInstanceOf[Array[Int]]
- val a3 = ColumnVectorUtils.toPrimitiveJavaArray(column.getArray(2)).asInstanceOf[Array[Int]]
- val a4 = ColumnVectorUtils.toPrimitiveJavaArray(column.getArray(3)).asInstanceOf[Array[Int]]
+ val a1 = ColumnVectorUtils.toJavaIntArray(column.getArray(0))
+ val a2 = ColumnVectorUtils.toJavaIntArray(column.getArray(1))
+ val a3 = ColumnVectorUtils.toJavaIntArray(column.getArray(2))
+ val a4 = ColumnVectorUtils.toJavaIntArray(column.getArray(3))
assert(a1 === Array(0))
assert(a2 === Array(1, 2))
assert(a3 === Array.empty[Int])
assert(a4 === Array(3, 4, 5))
// Verify the ArrayData APIs
- assert(column.getArray(0).length == 1)
+ assert(column.getArray(0).numElements() == 1)
assert(column.getArray(0).getInt(0) == 0)
- assert(column.getArray(1).length == 2)
+ assert(column.getArray(1).numElements() == 2)
assert(column.getArray(1).getInt(0) == 1)
assert(column.getArray(1).getInt(1) == 2)
- assert(column.getArray(2).length == 0)
+ assert(column.getArray(2).numElements() == 0)
- assert(column.getArray(3).length == 3)
+ assert(column.getArray(3).numElements() == 3)
assert(column.getArray(3).getInt(0) == 3)
assert(column.getArray(3).getInt(1) == 4)
assert(column.getArray(3).getInt(2) == 5)
@@ -677,7 +677,7 @@ class ColumnarBatchSuite extends SparkFunSuite {
assert(data.capacity == array.length * 2)
data.putInts(0, array.length, array, 0)
column.putArray(0, 0, array.length)
- assert(ColumnVectorUtils.toPrimitiveJavaArray(column.getArray(0)).asInstanceOf[Array[Int]]
+ assert(ColumnVectorUtils.toJavaIntArray(column.getArray(0))
=== array)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org