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
address review comments
  • Loading branch information
@kiszk
kiszk committed Aug 7, 2017
commit dc331f138ee22f3567bfc8a6f5b1e39e7102affd
161 changes: 2 additions & 159 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 @@ -30,7 +30,7 @@
* A column backed by data compressed thru ColumnAccessor
* this is a wrapper to read compressed data for table cache
*/
public final class CachedBatchColumnVector extends ColumnVector {
public final class CachedBatchColumnVector extends ReadOnlyColumnVector {

// accessor for a column
private ColumnAccessor columnAccessor;
Expand All @@ -49,7 +49,7 @@ public final class CachedBatchColumnVector extends ColumnVector {


public CachedBatchColumnVector(byte[] buffer, int numRows, DataType type) {
super(numRows, DataTypes.NullType, MemoryMode.ON_HEAP);
super(numRows, type, MemoryMode.ON_HEAP);
initialize(buffer, type);
reserveInternal(numRows);
reset();
Expand Down Expand Up @@ -89,26 +89,6 @@ private void prepareAccess(int rowId) {
// APIs dealing with nulls
//

@Override
public void putNotNull(int rowId) {
throw new UnsupportedOperationException();
}

@Override
public void putNull(int rowId) {
throw new UnsupportedOperationException();
}

@Override
public void putNulls(int rowId, int count) {
throw new UnsupportedOperationException();
}

@Override
public void putNotNulls(int rowId, int count) {
throw new UnsupportedOperationException();
}

@Override
public boolean isNullAt(int rowId) {
prepareAccess(rowId);
Expand All @@ -119,16 +99,6 @@ public boolean isNullAt(int rowId) {
// APIs dealing with Booleans
//

@Override
public void putBoolean(int rowId, boolean value) {
throw new UnsupportedOperationException();
}

@Override
public void putBooleans(int rowId, int count, boolean value) {
throw new UnsupportedOperationException();
}

@Override
public boolean getBoolean(int rowId) {
Copy link
Contributor

@cloud-fan cloud-fan Jul 18, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do we support reading values in a random order? e.g. getBoolean(2), getBoolean(1), getBoolean(2)?

Copy link
Member Author

@kiszk kiszk Jul 18, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We do not support reading values in a random order. This is because implementation of CompressionScheme (e.g. IntDelta) supports only sequential access.

Copy link
Contributor

@cloud-fan cloud-fan Jul 18, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

then we should throw exception for this case instead of returning wrong result.

Copy link
Member Author

@kiszk kiszk Jul 18, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see. I will add code to track access order for each getter.

prepareAccess(rowId);
Expand All @@ -146,21 +116,6 @@ public boolean[] getBooleans(int rowId, int count) {
// APIs dealing with Bytes
//

@Override
public void putByte(int rowId, byte value) {
throw new UnsupportedOperationException();
}

@Override
public void putBytes(int rowId, int count, byte value) {
throw new UnsupportedOperationException();
}

@Override
public void putBytes(int rowId, int count, byte[] src, int srcIndex) {
throw new UnsupportedOperationException();
}

@Override
public byte getByte(int rowId) {
prepareAccess(rowId);
Expand All @@ -176,21 +131,6 @@ public byte[] getBytes(int rowId, int count) {
// APIs dealing with Shorts
//

@Override
public void putShort(int rowId, short value) {
throw new UnsupportedOperationException();
}

@Override
public void putShorts(int rowId, int count, short value) {
throw new UnsupportedOperationException();
}

@Override
public void putShorts(int rowId, int count, short[] src, int srcIndex) {
throw new UnsupportedOperationException();
}

@Override
public short getShort(int rowId) {
prepareAccess(rowId);
Expand All @@ -206,26 +146,6 @@ public short[] getShorts(int rowId, int count) {
// APIs dealing with Ints
//

@Override
public void putInt(int rowId, int value) {
throw new UnsupportedOperationException();
}

@Override
public void putInts(int rowId, int count, int value) {
throw new UnsupportedOperationException();
}

@Override
public void putInts(int rowId, int count, int[] src, int srcIndex) {
throw new UnsupportedOperationException();
}

@Override
public void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex) {
throw new UnsupportedOperationException();
}

@Override
public int getInt(int rowId) {
prepareAccess(rowId);
Expand All @@ -245,26 +165,6 @@ public int getDictId(int rowId) {
// APIs dealing with Longs
//

@Override
public void putLong(int rowId, long value) {
throw new UnsupportedOperationException();
}

@Override
public void putLongs(int rowId, int count, long value) {
throw new UnsupportedOperationException();
}

@Override
public void putLongs(int rowId, int count, long[] src, int srcIndex) {
throw new UnsupportedOperationException();
}

@Override
public void putLongsLittleEndian(int rowId, int count, byte[] src, int srcIndex) {
throw new UnsupportedOperationException();
}

@Override
public long getLong(int rowId) {
prepareAccess(rowId);
Expand All @@ -280,26 +180,6 @@ public long[] getLongs(int rowId, int count) {
// APIs dealing with floats
//

@Override
public void putFloat(int rowId, float value) {
throw new UnsupportedOperationException();
}

@Override
public void putFloats(int rowId, int count, float value) {
throw new UnsupportedOperationException();
}

@Override
public void putFloats(int rowId, int count, float[] src, int srcIndex) {
throw new UnsupportedOperationException();
}

@Override
public void putFloats(int rowId, int count, byte[] src, int srcIndex) {
throw new UnsupportedOperationException();
}

@Override
public float getFloat(int rowId) {
prepareAccess(rowId);
Expand All @@ -315,26 +195,6 @@ public float[] getFloats(int rowId, int count) {
// APIs dealing with doubles
//

@Override
public void putDouble(int rowId, double value) {
throw new UnsupportedOperationException();
}

@Override
public void putDoubles(int rowId, int count, double value) {
throw new UnsupportedOperationException();
}

@Override
public void putDoubles(int rowId, int count, double[] src, int srcIndex) {
throw new UnsupportedOperationException();
}

@Override
public void putDoubles(int rowId, int count, byte[] src, int srcIndex) {
throw new UnsupportedOperationException();
}

@Override
public double getDouble(int rowId) {
prepareAccess(rowId);
Expand All @@ -359,11 +219,6 @@ public int getArrayOffset(int rowId) {
throw new UnsupportedOperationException();
}

@Override
public void putArray(int rowId, int offset, int length) {
throw new UnsupportedOperationException();
}

@Override
public void loadBytes(ColumnVector.Array array) {
throw new UnsupportedOperationException();
Expand All @@ -373,24 +228,12 @@ public void loadBytes(ColumnVector.Array array) {
// APIs dealing with Byte Arrays
//

@Override
public int putByteArray(int rowId, byte[] value, int offset, int length) {
throw new UnsupportedOperationException();
}

public final UTF8String getUTF8String(int rowId) {
prepareAccess(rowId);
return unsafeRow.getUTF8String(ORDINAL);
}

@Override
protected void reserveInternal(int newCapacity) {
capacity = newCapacity;
}

private void initialize(byte[] buffer, DataType type) {
this.type = type;

if (columnAccessor == null) {
ByteBuffer byteBuffer = ByteBuffer.wrap(buffer);
columnAccessor = ColumnAccessor$.MODULE$.apply(type, byteBuffer);
Expand Down
Loading