Skip to content

Commit 1d037b8

Browse files
committed
Fixed SPARK-1436: in-memory column byte buffer must be able to be accessed multiple times
Forgot to duplicate the in-memory column byte buffer when creating new ColumnAccessor's, so that when the column byte buffer is accessed multiple times, the position is not reset to 0.
1 parent d7c0e8f commit 1d037b8

File tree

2 files changed

+20
-11
lines changed

2 files changed

+20
-11
lines changed

sql/core/src/main/scala/org/apache/spark/sql/columnar/ColumnAccessor.scala

Lines changed: 12 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -100,20 +100,21 @@ private[sql] class GenericColumnAccessor(buffer: ByteBuffer)
100100

101101
private[sql] object ColumnAccessor {
102102
def apply(buffer: ByteBuffer): ColumnAccessor = {
103+
val dup = buffer.duplicate().order(ByteOrder.nativeOrder)
103104
// The first 4 bytes in the buffer indicate the column type.
104-
val columnTypeId = buffer.getInt()
105+
val columnTypeId = dup.getInt()
105106

106107
columnTypeId match {
107-
case INT.typeId => new IntColumnAccessor(buffer)
108-
case LONG.typeId => new LongColumnAccessor(buffer)
109-
case FLOAT.typeId => new FloatColumnAccessor(buffer)
110-
case DOUBLE.typeId => new DoubleColumnAccessor(buffer)
111-
case BOOLEAN.typeId => new BooleanColumnAccessor(buffer)
112-
case BYTE.typeId => new ByteColumnAccessor(buffer)
113-
case SHORT.typeId => new ShortColumnAccessor(buffer)
114-
case STRING.typeId => new StringColumnAccessor(buffer)
115-
case BINARY.typeId => new BinaryColumnAccessor(buffer)
116-
case GENERIC.typeId => new GenericColumnAccessor(buffer)
108+
case INT.typeId => new IntColumnAccessor(dup)
109+
case LONG.typeId => new LongColumnAccessor(dup)
110+
case FLOAT.typeId => new FloatColumnAccessor(dup)
111+
case DOUBLE.typeId => new DoubleColumnAccessor(dup)
112+
case BOOLEAN.typeId => new BooleanColumnAccessor(dup)
113+
case BYTE.typeId => new ByteColumnAccessor(dup)
114+
case SHORT.typeId => new ShortColumnAccessor(dup)
115+
case STRING.typeId => new StringColumnAccessor(dup)
116+
case BINARY.typeId => new BinaryColumnAccessor(dup)
117+
case GENERIC.typeId => new GenericColumnAccessor(dup)
117118
}
118119
}
119120
}

sql/core/src/test/scala/org/apache/spark/sql/columnar/ColumnarQuerySuite.scala

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,4 +31,12 @@ class ColumnarQuerySuite extends QueryTest {
3131

3232
checkAnswer(scan, testData.collect().toSeq)
3333
}
34+
35+
test("SPARK-1436 regression: in-memory columns must be able to be accessed multiple times") {
36+
val plan = TestSQLContext.executePlan(testData.logicalPlan).executedPlan
37+
val scan = SparkLogicalPlan(InMemoryColumnarTableScan(plan.output, plan))
38+
39+
checkAnswer(scan, testData.collect().toSeq)
40+
checkAnswer(scan, testData.collect().toSeq)
41+
}
3442
}

0 commit comments

Comments
 (0)