[SPARK-20783][SQL] Enhance ColumnVector to keep UnsafeArrayData for array #18014

kiszk · 2017-05-17T08:57:59Z

What changes were proposed in this pull request?

This PR enhances ColumnVector to keep UnsafeArrayData for array to use ColumnVector for table cache (e.g. CACHE table, DataFrame.cache). Other complex types such as Map and struct will be addressed by another PR if it is OK.

Current ColumnVector accepts only primitive-type Java array as an input for array. It is good to keep data from Parquet.

This PR changed or added the following APIs:

ColumnVector ColumnVector.allocate(int capacity, DataType type, MemoryMode mode, boolean useUnsafeArrayData)

When the last is true, the ColumnVector can keep UnsafeArrayData. If it is false, the ColumnVector cannot keep UnsafeArrayData.

int ColumnVector.putArray(int rowId, ArrayData array)

When this ColumnVector was generated with useUnsafeArrayData=true, this method stores UnsafeArrayData into ColumnVector. Otherwise, throw an exception.

ArrayData ColumnVector.getArray(int rowId)

When this ColumnVector was generated with useUnsafeArrayData=true, this method returns UnsafeArrayData.

How was this patch tested?

Update existing testsuite

kiszk · 2017-05-17T09:49:36Z

Jenkins, test this please

kiszk · 2017-05-17T16:59:05Z

Jenkins, test this please

SparkQA · 2017-05-17T17:01:57Z

Test build #77017 has finished for PR 18014 at commit b939a08.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-05-17T17:05:12Z

Test build #77023 has finished for PR 18014 at commit b939a08.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-05-17T19:20:37Z

Test build #77024 has finished for PR 18014 at commit 3f726ba.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

kiszk · 2017-05-17T19:37:35Z

@hvanhovell @sameeragarwal would it be possible to look at this?
cc: @cloud-fan

cloud-fan · 2017-05-18T04:34:56Z

I may miss something, can we just treat array type as binary type and put it in ColumnVector?

kiszk · 2017-05-18T05:21:15Z

I thought that idea is for Apache Arrow.
We could use binary type for UnsafeArrayData. However, it involves some complexity to use ColumnVector.Array. I wanted to simply a path by avoiding ColumnVector.Array. I think that this is easier to understand.

Is it better to use existing code?

kiszk · 2017-05-19T12:01:08Z

@cloud-fan What would you think?

kiszk · 2017-05-22T22:30:40Z

@cloud-fan Could you please let us know your thoughts?
Is it better to use binary type or to add simple logic for UnsafeArrayData and others in ColumnVector?

cloud-fan · 2017-05-23T08:25:43Z

I think there is a gap between columnar format and the unsafe row format. The current ColumnVector format looks reasonable for array type, as it puts the leaf elements together, which is better for compression.

While changing it to binary makes it faster if we need to read an array from ColumnVector and convert it to ArrayData, it's really an inefficient way to store arrays in columnar format. Shall we consider updatingUnsafeArrayData format?

kiszk · 2017-05-23T12:54:35Z

@cloud-fan Thank you for your comments. Let me confirm your ideas.

Do you want to keep array contents in a primitive data array (e.g. intData[]) of UnsafeArrayData?
How do you want to update UnsafeArrayData?

We can map the current UnsafeArrayData into ColumnVector. The following is the format of UnsafeArrayData.

 [numElements][null bits][values or offset&length][variable length portion]

numElements: store it into arrayLengths[]
[null bits]: Need to conversion from bitvector representation to byte representation
[values]: store as each data type
[offset&length][variable length portion]: store as ByteType

The issue is a conversion of null bits. However, if we use byte representation in UnsafeArrayData, it may waste more memory space. To avoid this, we could update ColumnVector to support bitvector representation for nullability of each element.

On the other hand, current my approach stores the whole UnsafeArrayData as a binary into byte[] data. The advantage of this approach is no cost for conversion at put/get.

What do you think?

cloud-fan · 2017-05-24T09:36:30Z

I took a look at ColumnVector.getArray, seems it's already no cost? The writing needs some copy though.

kiszk · 2017-05-24T19:19:06Z

Yes. Let me implement new putArray(int rowId, Array array) that uses ColumnVector.Array and stores primitive-type elements into a primitive array (e.g. intData).

kiszk · 2017-05-25T00:24:08Z

@cloud-fan When I think about the use case of ColumnVector.getArray (i.e. in generated code by the whole-stage code geneneration), I think that it is better to return UnsafeArrayData instead of ColumnVector.Array in the currnet generated code.

The following is an example program and current generated code. In the generated code, we will replace inputadapter_row.getArray(0) at line 35 with columnVector.getArray(rowIdx). So, let us assume columnVector.getArray(rowIdx) is used and focus on inputadapter_value.

At projection, if a type of inputadapter_value is UnsafeArrayData, line 72 just performs fast memory copy for a contigious region. On the other hand, if a type of inputadapter_value is ColumnVector.Array, lines 79-86 performs element-wise copy that is slower.

I think that there are three options.

Add a method UnsafeArrayData ColumnVector.getArray() to use UnsafeArrayData anywhere in the generated code.
Add a conditional branch for ColumnVector.Array in the generated code and prepare specialized copy routine for ColumnVector.Array. In this case, we can use bulk copy for array body, but use element-wise copy for null bits.
In addition to 2, we support bit-wise null bits representation in ColumnVector.Array. In this case, we can use two bulk copy. One is for null bits, and the other is for array body. Potential cons is to introduce additional conditional branch in isNullAt() for checking whether bit-wise or byte-wise representation is used.

What do you think? Or, are there any other ideas?

val df = sparkContext.parallelize(Seq(Array(0, 1), Array(1, 2)), 1).toDF("a").cache
df.count
df.filter("a[0] > 0").show

/* 031 */   protected void processNext() throws java.io.IOException {
/* 032 */     while (inputadapter_input.hasNext() && !stopEarly()) {
/* 033 */       InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();
/* 034 */       boolean inputadapter_isNull = inputadapter_row.isNullAt(0);
/* 035 */       ArrayData inputadapter_value = inputadapter_isNull ? null : (inputadapter_row.getArray(0));
/* 037 */       if (!(!(inputadapter_isNull))) continue;
/* 039 */       boolean filter_isNull2 = true;
/* 040 */       boolean filter_value2 = false;
/* 042 */       boolean filter_isNull3 = true;
/* 043 */       int filter_value3 = -1;
/* 045 */       filter_isNull3 = false;
/* 047 */       final int filter_index = (int) 0;
/* 048 */       if (filter_index >= inputadapter_value.numElements() || filter_index < 0 || inputadapter_value.isNullAt(filter_index)) {
/* 049 */         filter_isNull3 = true;
/* 050 */       } else {
/* 051 */         filter_value3 = inputadapter_value.getInt(filter_index);
/* 052 */       }
/* 053 */       if (!filter_isNull3) {
/* 054 */         filter_isNull2 = false;
/* 055 */         filter_value2 = filter_value3 > 0;
/* 057 */       }
/* 058 */       if (filter_isNull2 || !filter_value2) continue;
/* 060 */       filter_numOutputRows.add(1);
/* 062 */       filter_holder.reset();
/* 066 */       final int filter_tmpCursor = filter_holder.cursor;
/* 068 */       if (inputadapter_value instanceof UnsafeArrayData) {
/* 069 */         final int filter_sizeInBytes = ((UnsafeArrayData) inputadapter_value).getSizeInBytes();
/* 071 */         filter_holder.grow(filter_sizeInBytes);
/* 072 */         ((UnsafeArrayData) inputadapter_value).writeToMemory(filter_holder.buffer, filter_holder.cursor);
/* 073 */         filter_holder.cursor += filter_sizeInBytes;
/* 075 */       } else {
/* 076 */         final int filter_numElements = inputadapter_value.numElements();
/* 077 */         filter_arrayWriter.initialize(filter_holder, filter_numElements, 4);
/* 079 */         for (int filter_index1 = 0; filter_index1 < filter_numElements; filter_index1++) {
/* 080 */           if (inputadapter_value.isNullAt(filter_index1)) {
/* 081 */             filter_arrayWriter.setNullInt(filter_index1);
/* 082 */           } else {
/* 083 */             final int filter_element = inputadapter_value.getInt(filter_index1);
/* 084 */             filter_arrayWriter.write(filter_index1, filter_element);
/* 085 */           }
/* 086 */         }
/* 087 */       }
/* 089 */       filter_rowWriter.setOffsetAndSize(0, filter_tmpCursor, filter_holder.cursor - filter_tmpCursor);
/* 090 */       filter_result.setTotalSize(filter_holder.totalSize());
/* 091 */       append(filter_result);
/* 092 */       if (shouldStop()) return;
/* 093 */     }
/* 094 */   }

cloud-fan · 2017-05-25T12:35:59Z

I think option 2 is better. ColumnVector.getArray() should be as fast as possible. The caller side may just get an element from this array and then the final projection doesn't need to copy an array.

But if we do need to copy the array in the final projection, we should also speed it up by bulk copy.

For the write path, ColumnVector.putArray, we should also avoid element-wise copy and prefer bulk copy.

SparkQA · 2017-05-25T19:49:48Z

Test build #77384 has finished for PR 18014 at commit bf6ab20.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-05-25T22:07:00Z

Test build #77385 has finished for PR 18014 at commit 9954d6b.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

kiszk · 2017-05-26T01:42:24Z

@cloud-fan Here is an implementation based on Option 2 only for simple data types (e.g. boolean, int, double, and so on). Used bulk-copy for array body in putArray(), and used element-wise copy for null bit if containsNull() is true in putArray().
Do you have any comments/feedbacks?

cloud-fan · 2017-05-26T01:59:33Z

sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java

+    } else if (et == DataTypes.DoubleType) {
+      Platform.copyMemory(
+        src, srcOffset, doubleData, Platform.DOUBLE_ARRAY_OFFSET + dstOffset * 8, numElements * 8);
+    } else {


if we wanna support nested arrays, do we need to do multiple bulk copies?

I think we may need to update UnsafeArrayData to put leaf elements together like ColumnVector did, then we can just use one bulk copy to write a nested array.

You are right, current implementation will require do multiple bulk copies for a nested array.

IIUC, UnsafeArrayData put all of the elements together in a contiguous memory region even when it is a nested array.
On the other hand, current ColumnVector puts elements for each dimension in different ColumnVector.childColumns[0]. Thus, multiple bulk copies are required.
If there is an array Array(Array(1), Array(2)), in ColumnVector, Array(1) and Array(2) are stored into different intData in ColumnVector.childColumns[0].
Am I correct?

no I think it's the opposite. UnsafeArrayData will blindly put elements together, if the element is also array type, each element will contain null bits, so the leaf elements are not together. For ColumnVector, you can take a look at ColumnarBatchSuite.String APIs, the leaf elements are together

ping @cloud-fan

I don't think so, ColumnVector.Array.getArray actually return the same instance.

Let me think about this over weekend since I am busy to prepare a slide for SparkSummit.

@cloud-fan I agree that ColumnVector.Array.getArray returns the same instance in ColumnVector.resultArray. This is good for one-dimensional array.

In my understanding, we have to get ColumnVector from this instance to supporting nested array. However, current ColumnVector.Array does not have getter for ColumnVector. It is hard to implement nested array in current ColumnVector implementation. Am I missing something? Or, do you have any idea to enhance ColumnVector?

ping @cloud-fan

gatorsmile · 2017-10-28T05:34:56Z

What is the latest status of this PR?

kiszk · 2017-10-28T08:03:45Z

It is older one. Let me close this. Then, I will submit another PR very soon to do the same thing in different way.

Keep UnsafeArrayData for Array in ColumnVector

b939a08

fix scala style

3f726ba

kiszk mentioned this pull request May 21, 2017

[SPARK-17915][SQL] Prepare a new ColumnVector implementation for UnsafeData #15468

Closed

use ColumnVector.Array

bf6ab20

fix scala style error

9954d6b

cloud-fan reviewed May 26, 2017

View reviewed changes

kiszk closed this Oct 28, 2017

[SPARK-20783][SQL] Enhance ColumnVector to keep UnsafeArrayData for array #18014

[SPARK-20783][SQL] Enhance ColumnVector to keep UnsafeArrayData for array #18014

Uh oh!

Conversation

kiszk commented May 17, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

kiszk commented May 17, 2017

Uh oh!

kiszk commented May 17, 2017

Uh oh!

SparkQA commented May 17, 2017

Uh oh!

SparkQA commented May 17, 2017

Uh oh!

SparkQA commented May 17, 2017

Uh oh!

kiszk commented May 17, 2017

Uh oh!

cloud-fan commented May 18, 2017

Uh oh!

kiszk commented May 18, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

kiszk commented May 19, 2017

Uh oh!

kiszk commented May 22, 2017

Uh oh!

cloud-fan commented May 23, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

kiszk commented May 23, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cloud-fan commented May 24, 2017

Uh oh!

kiszk commented May 24, 2017

Uh oh!

kiszk commented May 25, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cloud-fan commented May 25, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented May 25, 2017

Uh oh!

SparkQA commented May 25, 2017

Uh oh!

kiszk commented May 26, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan May 26, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kiszk Jun 6, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gatorsmile commented Oct 28, 2017

Uh oh!

kiszk commented Oct 28, 2017

Uh oh!

kiszk commented May 17, 2017 •

edited

Loading

kiszk commented May 18, 2017 •

edited

Loading

cloud-fan commented May 23, 2017 •

edited

Loading

kiszk commented May 23, 2017 •

edited

Loading

kiszk commented May 25, 2017 •

edited

Loading

cloud-fan commented May 25, 2017 •

edited

Loading

cloud-fan May 26, 2017 •

edited

Loading

kiszk Jun 6, 2017 •

edited

Loading