[SPARK-21046][SQL] simplify the array offset and length in ColumnVector #18260

cloud-fan · 2017-06-10T09:45:38Z

What changes were proposed in this pull request?

Currently when a ColumnVector stores array type elements, we will use 2 arrays for lengths and offsets and implement them individually in on-heap and off-heap column vector.

In this PR, we use one array to represent both offsets and lengths, so that we can treat it as ColumnVector and all the logic can go to the base class ColumnVector

How was this patch tested?

existing tests.

cloud-fan · 2017-06-10T09:45:53Z

cc @ueshin @sameeragarwal @kiszk

kiszk · 2017-06-10T09:56:53Z

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

-      if (this.arrayLengths != null) {
-        System.arraycopy(this.arrayLengths, 0, newLengths, 0, capacity);
-        System.arraycopy(this.arrayOffsets, 0, newOffsets, 0, capacity);
+      // need 2 ints as offset and length for each array.


It would be good to add a comment regarding that intData[] is used for other purpose compared to the original intention (i.e. storing int column data).

yea good idea.

kiszk · 2017-06-10T10:02:30Z

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

-   * Returns the offset of the array at rowid.
-   */
-  public abstract int getArrayOffset(int rowId);
+  public void arrayWriteEnd(int rowId, int offset, int length) {


I assumed that there is arrayWriteStart since we newly created arrayWriteEnd.
After reading a comment, I understood your intention. Is there any other method name (e.g. arraySetColumn)?

no there is no arrayWriteStart.

Maybe I should pick another name, how about putArrayOffsetAndLength or just keep the original name?

I like putArrayOffsetAndLength

SparkQA · 2017-06-10T12:23:05Z

Test build #77874 has finished for PR 18260 at commit d4267b7.

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

ueshin · 2017-06-12T00:23:13Z

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

  // The data stored in these two allocations need to maintain binary compatible. We can
  // directly pass this buffer to external components.
  private long nulls;
+  // The actually data of this column vector will be store here. If it's an array column vector,


nit: will be store -> will be stored ?

ueshin · 2017-06-12T00:23:59Z

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

  // Array for each type. Only 1 is populated for any type.
  private byte[] byteData;
  private short[] shortData;
+  // This is not used used to store data for int column vector, but also can store offsets and


nit: This is not used used to store -> This is not used to store ?

ueshin · 2017-06-12T00:27:11Z

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

-          Platform.reallocateMemory(offsetData, oldCapacity * 4, newCapacity * 4);
+      // need 2 ints as offset and length for each array.
+      this.data = Platform.reallocateMemory(data, oldCapacity * 8, newCapacity * 8);
+      putInt(0, 0);


Do we need to putInt here?

actually we don't, as the new memory region should be filled with 0. This is just a safe guard to be more robust about the underlying memory allocation details.

I see, thanks.

I had second thought that putArrayOffsetAndLength(0,0,0) is better for the purpose of the guard?

ueshin · 2017-06-12T00:37:15Z

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

      }
-      arrayLengths = newLengths;
-      arrayOffsets = newOffsets;
+      putInt(0, 0);


ueshin · 2017-06-12T00:44:43Z

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

+      // need 2 ints as offset and length for each array.
+      if (intData == null || intData.length < newCapacity * 2) {
+        int[] newData = new int[newCapacity * 2];
+        if (intData != null) System.arraycopy(intData, 0, newData, 0, capacity * 2);


~~I guess we should pass intData.length instead of capacity * 2 ?~~
nvm: here intData.length should be capacity * 2.

good catch!

ueshin · 2017-06-12T01:37:59Z

sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala

-      column.putArray(3, 3, 3)
+      column.putArrayOffsetAndLength(0, 0, 1)
+      column.putArrayOffsetAndLength(1, 1, 2)
+      column.putArrayOffsetAndLength(2, 2, 0)


This should be column.putArrayOffsetAndLength(2, 3, 0) ?

ueshin · 2017-06-12T01:50:01Z

LGTM except for one comment, pending Jenkins.

viirya · 2017-06-12T02:59:47Z

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

-          Platform.reallocateMemory(lengthData, oldCapacity * 4, newCapacity * 4);
-      this.offsetData =
-          Platform.reallocateMemory(offsetData, oldCapacity * 4, newCapacity * 4);
+      // need 2 ints as offset and length for each array.


nit: for each array -> for each array element.

Oh. This's quite a bit ambiguous. Nvm. for each array is good.

SparkQA · 2017-06-12T03:24:21Z

Test build #77902 has finished for PR 18260 at commit 2b78043.

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

viirya · 2017-06-12T03:37:36Z

LGTM except for one question regarding array capacity after this change.

viirya · 2017-06-12T03:48:16Z

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

  private short[] shortData;
+  // This is not only used to store data for int column vector, but also can store offsets and
+  // lengths for array column vector.
  private int[] intData;


One question I just have is, the capacity of ColumnVector is bound by MAX_CAPACITY. Previously we store offset and length individually, so we can have MAX_CAPACITY arrays at most. Now we store offset and length together in data/intData which is bound to MAX_CAPACITY, doesn't it say we can just have MAX_CAPACITY / 2 arrays at most?

Good catch. Is it possible to use longData, which has a pair of 32-bit offset and length, to keep MAX_CAPACITY array length?

~~Oh. I see. We only check the limit of MAX_CAPACITY before actually going into reserveInternal.~~

Even we pass this check, we still face problem when allocating intData, see the comment below.

@kiszk Do you meant we store a pair of offset/length together as an element in longData?

SparkQA · 2017-06-12T03:56:28Z

Test build #77903 has finished for PR 18260 at commit 368c346.

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

viirya · 2017-06-12T04:08:17Z

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

-        System.arraycopy(this.arrayOffsets, 0, newOffsets, 0, capacity);
+      // need 2 ints as offset and length for each array.
+      if (intData == null || intData.length < newCapacity * 2) {
+        int[] newData = new int[newCapacity * 2];


newCapacity here can be MAX_CAPACITY at most. When newCapacity is more than MAX_CAPACITY / 2, seems this allocation would cause problem?

SparkQA · 2017-06-12T04:33:34Z

Test build #77900 has finished for PR 18260 at commit a61ba71.

This patch fails from timeout after a configured wait of `250m`.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-06-12T04:36:24Z

Test build #77905 has finished for PR 18260 at commit 1dae660.

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

cloud-fan · 2017-06-12T05:07:42Z

@viirya @kiszk good catch! fixed by using long to store offset and size.

viirya · 2017-06-12T05:15:56Z

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

  private byte[] byteData;
  private short[] shortData;
  private int[] intData;
+  // This is not only used to store data for int column vector, but also can store offsets and


int column vector -> long column vector.

viirya · 2017-06-12T05:21:07Z

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

-   */
-  public abstract int getArrayOffset(int rowId);
+  public void putArrayOffsetAndSize(int rowId, int offset, int size) {
+    long offsetAndSize = (offset << 32) | size;


offset should be converted to long before shifting?

viirya · 2017-06-12T05:23:55Z

LGTM except for two comments.

SparkQA · 2017-06-12T06:29:03Z

Test build #77912 has finished for PR 18260 at commit e6e60e0.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-06-12T13:19:06Z

Test build #77933 has finished for PR 18260 at commit fdc0870.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2017-06-12T13:30:09Z

sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchSuite.scala


      reference.zipWithIndex.foreach { v =>
-        assert(v._1.length == column.getArrayLength(v._2), "MemoryMode=" + memMode)
+        assert(v._1.length == column.getInt(v._2 * 2 + 1), "MemoryMode=" + memMode)


ah, we should also change here.

SparkQA · 2017-06-12T16:06:23Z

Test build #77936 has finished for PR 18260 at commit 04790bb.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-06-12T16:11:52Z

The spark R test failure is not related, I'm merging it to master, thanks for your review!

rxin · 2017-06-12T17:10:25Z

Why are we doing this? Isn't it better potentially for compression to store them separately? We can also easily remove the offset for fixed length arrays.

cloud-fan · 2017-06-13T01:14:23Z

Sorry I missed this part. Currently we don't externalize ColumnVector so I didn't go with this direction. I'm reverting it, waiting to have a consensus about how to change the ColumnVector APIs.

## What changes were proposed in this pull request? Currently when a `ColumnVector` stores array type elements, we will use 2 arrays for lengths and offsets and implement them individually in on-heap and off-heap column vector. In this PR, we use one array to represent both offsets and lengths, so that we can treat it as `ColumnVector` and all the logic can go to the base class `ColumnVector` ## How was this patch tested? existing tests. Author: Wenchen Fan <[email protected]> Closes apache#18260 from cloud-fan/put.

simplify the array offset and length in ColumnVector

d4267b7

kiszk reviewed Jun 10, 2017

View reviewed changes

address comments

a61ba71

ueshin reviewed Jun 12, 2017

View reviewed changes

more comments

368c346

cloud-fan force-pushed the put branch from 2b78043 to 368c346 Compare June 12, 2017 01:21

ueshin reviewed Jun 12, 2017

View reviewed changes

minor

1dae660

viirya reviewed Jun 12, 2017

View reviewed changes

use long array

e6e60e0

viirya reviewed Jun 12, 2017

View reviewed changes

fix bug

fdc0870

viirya reviewed Jun 12, 2017

View reviewed changes

fix test

04790bb

asfgit closed this in 22dd65f Jun 12, 2017

[SPARK-21046][SQL] simplify the array offset and length in ColumnVector #18260

[SPARK-21046][SQL] simplify the array offset and length in ColumnVector #18260

Uh oh!

Conversation

cloud-fan commented Jun 10, 2017

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

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

Uh oh!

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jun 10, 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!

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!

ueshin Jun 12, 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!

ueshin commented Jun 12, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya Jun 12, 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!

SparkQA commented Jun 12, 2017

Uh oh!

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

Uh oh!

Uh oh!

viirya Jun 12, 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!

viirya Jun 12, 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!

SparkQA commented Jun 12, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jun 12, 2017

Uh oh!

SparkQA commented Jun 12, 2017

Uh oh!

cloud-fan commented Jun 10, 2017 •

edited

Loading

ueshin Jun 12, 2017 •

edited

Loading

viirya Jun 12, 2017 •

edited

Loading

viirya commented Jun 12, 2017 •

edited

Loading

viirya Jun 12, 2017 •

edited

Loading

viirya Jun 12, 2017 •

edited

Loading

cloud-fan commented Jun 12, 2017 •

edited

Loading

cloud-fan commented Jun 13, 2017 •

edited

Loading