Skip to content

[SPARK-20783][SQL] Create CachedBatchColumnVector to abstract existing compressed column #18468

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 21 commits into from
Closed

[SPARK-20783][SQL] Create CachedBatchColumnVector to abstract existing compressed column #18468

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
d4c7d1a
initial commit
kiszk Jun 29, 2017
176fcc9
fix scala style error
kiszk Jun 29, 2017
920b38b
introduced VectorType into ColumnVector
kiszk Jul 7, 2017
aa67f9c
add a new file
kiszk Jul 8, 2017
de27a10
remove ON_HEAP_CACHEDBATCH
kiszk Jul 8, 2017
bb47299
add missing methods
kiszk Jul 8, 2017
cb3e631
add ColumnAccessor.extract(ColumnVector, rowId)
kiszk Jul 11, 2017
68f7e39
rename OnHeapCachedBatch.java to CachedBatchColumnVector.java
kiszk Jul 18, 2017
a068ada
address review comments
kiszk Jul 18, 2017
7c13ed6
address review comments
kiszk Jul 18, 2017
d8fede6
address review comments
kiszk Jul 19, 2017
780115b
address review comment
kiszk Jul 19, 2017
31d2dd7
address review comments
kiszk Jul 19, 2017
8aa9b16
address review comments
kiszk Jul 19, 2017
332fc6c
address review comment
kiszk Jul 20, 2017
dc331f1
address review comments
kiszk Jul 20, 2017
70042d5
fix test faiures
kiszk Jul 21, 2017
d16779f
address review comment
kiszk Jul 21, 2017
ff324fd
addressed review comments
kiszk Jul 21, 2017
157fde1
add initialize() method for reusing CachedBatchColumnVector
kiszk Aug 1, 2017
a26dc15
make initialize public
kiszk Aug 6, 2017
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
addressed review comments
  • Loading branch information
@kiszk
kiszk committed Aug 7, 2017
commit ff324fd9f588ace1f4f16d62b71d2de6842de102
20 changes: 8 additions & 12 deletions ...core/src/main/java/org/apache/spark/sql/execution/vectorized/CachedBatchColumnVector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@
import org.apache.spark.unsafe.types.UTF8String;

/**
* A column backed by data compressed thru ColumnAccessor
* A column vector backed by data compressed thru ColumnAccessor
* this is a wrapper to read compressed data for table cache
*/
public final class CachedBatchColumnVector extends ReadOnlyColumnVector {
Expand All @@ -51,7 +51,6 @@ public final class CachedBatchColumnVector extends ReadOnlyColumnVector {
public CachedBatchColumnVector(byte[] buffer, int numRows, DataType type) {
super(numRows, type, MemoryMode.ON_HEAP);
initialize(buffer, type);
reset();
}

@Override
Expand Down Expand Up @@ -233,16 +232,13 @@ public final UTF8String getUTF8String(int rowId) {
}

private void initialize(byte[] buffer, DataType type) {
if (columnAccessor == null) {
ByteBuffer byteBuffer = ByteBuffer.wrap(buffer);
columnAccessor = ColumnAccessor$.MODULE$.apply(type, byteBuffer);
}
if (mutableRow == null) {
unsafeRow = new UnsafeRow(1);
bufferHolder = new BufferHolder(unsafeRow);
rowWriter = new UnsafeRowWriter(bufferHolder, 1);
mutableRow = new MutableUnsafeRow(rowWriter);
}
ByteBuffer byteBuffer = ByteBuffer.wrap(buffer);
columnAccessor = ColumnAccessor$.MODULE$.apply(type, byteBuffer);

unsafeRow = new UnsafeRow(1);
bufferHolder = new BufferHolder(unsafeRow);
rowWriter = new UnsafeRowWriter(bufferHolder, 1);
mutableRow = new MutableUnsafeRow(rowWriter);

if (type instanceof ArrayType) {
throw new UnsupportedOperationException();
Expand Down
44 changes: 18 additions & 26 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1257,9 +1257,8 @@ class ColumnarBatchSuite extends SparkFunSuite {
val columnBuilder = ColumnBuilderHelper(dataType, 1024, "col", true)
val row = new SpecificInternalRow(Array(dataType))

val nullRow = InternalRow(false)
nullRow.setNullAt(0)
columnBuilder.appendFrom(nullRow, 0)
row.setNullAt(0)
columnBuilder.appendFrom(row, 0)
for (i <- 1 until 16) {
row.setBoolean(0, i % 2 == 0)
columnBuilder.appendFrom(row, 0)
Expand All @@ -1281,9 +1280,8 @@ class ColumnarBatchSuite extends SparkFunSuite {
val columnBuilder = ColumnBuilderHelper(dataType, 1024, "col", true)
val row = new SpecificInternalRow(Array(dataType))

val nullRow = InternalRow(false)
nullRow.setNullAt(0)
columnBuilder.appendFrom(nullRow, 0)
row.setNullAt(0)
columnBuilder.appendFrom(row, 0)
for (i <- 1 until 16) {
row.setByte(0, i.toByte)
columnBuilder.appendFrom(row, 0)
Expand All @@ -1305,9 +1303,8 @@ class ColumnarBatchSuite extends SparkFunSuite {
val columnBuilder = ColumnBuilderHelper(dataType, 1024, "col", true)
val row = new SpecificInternalRow(Array(dataType))

val nullRow = InternalRow(false)
nullRow.setNullAt(0)
columnBuilder.appendFrom(nullRow, 0)
row.setNullAt(0)
columnBuilder.appendFrom(row, 0)
for (i <- 1 until 16) {
row.setShort(0, i.toShort)
columnBuilder.appendFrom(row, 0)
Expand All @@ -1329,9 +1326,8 @@ class ColumnarBatchSuite extends SparkFunSuite {
val columnBuilder = ColumnBuilderHelper(dataType, 1024, "col", true)
val row = new SpecificInternalRow(Array(dataType))

val nullRow = InternalRow(false)
nullRow.setNullAt(0)
columnBuilder.appendFrom(nullRow, 0)
row.setNullAt(0)
columnBuilder.appendFrom(row, 0)
for (i <- 1 until 16) {
row.setInt(0, i)
columnBuilder.appendFrom(row, 0)
Expand All @@ -1353,9 +1349,8 @@ class ColumnarBatchSuite extends SparkFunSuite {
val columnBuilder = ColumnBuilderHelper(dataType, 1024, "col", true)
val row = new SpecificInternalRow(Array(dataType))

val nullRow = InternalRow(false)
nullRow.setNullAt(0)
columnBuilder.appendFrom(nullRow, 0)
row.setNullAt(0)
columnBuilder.appendFrom(row, 0)
for (i <- 1 until 16) {
row.setLong(0, i.toLong)
columnBuilder.appendFrom(row, 0)
Expand All @@ -1377,9 +1372,8 @@ class ColumnarBatchSuite extends SparkFunSuite {
val columnBuilder = ColumnBuilderHelper(dataType, 1024, "col", true)
val row = new SpecificInternalRow(Array(dataType))

val nullRow = InternalRow(false)
nullRow.setNullAt(0)
columnBuilder.appendFrom(nullRow, 0)
row.setNullAt(0)
columnBuilder.appendFrom(row, 0)
for (i <- 1 until 16) {
row.setFloat(0, i.toFloat)
columnBuilder.appendFrom(row, 0)
Expand All @@ -1401,9 +1395,8 @@ class ColumnarBatchSuite extends SparkFunSuite {
val columnBuilder = ColumnBuilderHelper(dataType, 1024, "col", true)
val row = new SpecificInternalRow(Array(dataType))

val nullRow = InternalRow(false)
nullRow.setNullAt(0)
columnBuilder.appendFrom(nullRow, 0)
row.setNullAt(0)
columnBuilder.appendFrom(row, 0)
for (i <- 1 until 16) {
row.setDouble(0, i.toDouble)
columnBuilder.appendFrom(row, 0)
Expand All @@ -1423,13 +1416,12 @@ class ColumnarBatchSuite extends SparkFunSuite {
test("CachedBatch String type Apis") {
val dataType = StringType
val columnBuilder = ColumnBuilderHelper(dataType, 1024, "col", true)
val row = new SpecificInternalRow(Array(dataType))

val nullRow = InternalRow("")
nullRow.setNullAt(0)
columnBuilder.appendFrom(nullRow, 0)
row.setNullAt(0)
columnBuilder.appendFrom(row, 0)
for (i <- 1 until 16) {
val converter = UnsafeProjection.create(Array[DataType](dataType))
val row = converter.apply(InternalRow(UTF8String.fromString((i % 4).toString)))
row.update(0, UTF8String.fromString((i % 4).toString))
columnBuilder.appendFrom(row, 0)
}

Expand Down