You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by nongli <gi...@git.apache.org> on 2016/01/07 05:53:40 UTC
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
GitHub user nongli opened a pull request:
https://github.com/apache/spark/pull/10628
[SPARK-12635][SQL] Add ColumnarBatch, an in memory columnar format for execution.
There are many potential benefits of having an efficient in memory columnar format as an alternate
to UnsafeRow. This patch introduces ColumnarBatch/ColumnarVector which starts this effort. The
remaining implementation can be done as follow up patches.
As stated in the in the JIRA, there are useful external components that operate on memory in a
simple columnar format. ColumnarBatch would serve that purpose and could server as a
zero-serialization/zero-copy exchange for this use case.
This patch supports running the underlying data either on heap or off heap. On heap runs a bit
faster but we would need offheap for zero-copy exchanges. Currently, this mode is hidden behind one
interface (ColumnVector).
This differs from Parquet or the existing columnar cache because this is *not* intended to be used
as a storage format. The focus is entirely on CPU efficiency as we expect to only have 1 of these
batches in memory per task. The layout of the values is just dense arrays of the value type.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/nongli/spark spark-12635
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/10628.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #10628
----
commit 381492eee072a8e1a536103db1e6afcff008b6b1
Author: Nong <no...@gmail.com>
Date: 2016-01-01T05:12:44Z
[SPARK-12635][SQL] Add ColumnarBatch, an in memory columnar format for execution.
There are many potential benefits of having an efficient in memory columnar format as an alternate
to UnsafeRow. This patch introduces ColumnarBatch/ColumnarVector which starts this effort. The
remaining implementation can be done as follow up patches.
As stated in the in the JIRA, there are useful external components that operate on memory in a
simple columnar format. ColumnarBatch would serve that purpose and could server as a
zero-serialization/zero-copy exchange for this use case.
This patch supports running the underlying data either on heap or off heap. On heap runs a bit
faster but we would need offheap for zero-copy exchanges. Currently, this mode is hidden behind one
interface (ColumnVector).
This differs from Parquet or the existing columnar cache because this is *not* intended to be used
as a storage format. The focus is entirely on CPU efficiency as we expect to only have 1 of these
batches in memory per task.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12785][SQL] Add ColumnarBatch, an in me...
Posted by robbinspg <gi...@git.apache.org>.
Github user robbinspg commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-204684533
So big endian implementations of OffHeapColumnVector and OnHeapColumnVector are needed. I don't think we'd want to have an inline 'if (bigEndian)' in the relevent methods so it may be that we'd want to subclass those classes, override the methods which require handling big endian, and instantiate the big endian version in ColumnVector.allocate if on a BE platform?
xxxHeapColumnVector classes would have to lose the 'final' attribute.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49376107
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java ---
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+
+/**
+ * This class represents a column of values and provides the main APIs to access the data
+ * values. It supports all the types and contains get/put APIs as well as their batched versions.
+ * The batched versions are preferable whenever possible.
+ *
+ * Most of the APIs take the rowId as a parameter. This is the local 0-based row id for values
+ * in the current RowBatch.
+ *
+ * A ColumnVector should be considered immutable once originally created. In other words, it is not
+ * valid to call put APIs after reads until reset() is called.
+ */
+public abstract class ColumnVector {
+ /**
+ * Allocates a column with each element of size `width` either on or off heap.
+ */
+ public static ColumnVector allocate(int capacity, DataType type, boolean offHeap) {
+ if (offHeap) {
+ return new OffHeapColumnVector(capacity, type);
+ } else {
+ return new OnHeapColumnVector(capacity, type);
+ }
+ }
+
+ public final DataType dataType() { return type; }
+
+ /**
+ * Resets this column for writing. The currently stored values are no longer accessible.
+ */
+ public void reset() {
+ numNulls = 0;
+ if (anyNullsSet) {
+ putNotNulls(0, capacity);
+ anyNullsSet = false;
+ }
+ }
+
+ /**
+ * Cleans up memory for this column. The column is not usable after this.
+ * TODO: this should probably have ref-counted semantics.
+ */
+ public abstract void close();
+
+ /**
+ * Returns the number of nulls in this column.
+ */
+ public final int numNulls() { return numNulls; }
+
+ /**
+ * Returns true if any of the nulls indicator are set for this column. This can be used
+ * as an optimization to prevent setting nulls.
+ */
+ public final boolean anyNullsSet() { return anyNullsSet; }
+
+ /**
+ * Returns the off heap ptr for the arrays backing the NULLs and values buffer. Only valid
+ * to call for off heap columns.
+ */
+ public abstract long nullsNativeAddress();
+ public abstract long valuesNativeAddress();
+
+ /**
+ * Sets the value at rowId to null/not null.
+ */
+ public abstract void putNotNull(int rowId);
+ public abstract void putNull(int rowId);
+
+ /**
+ * Sets the values from [rowId, rowId + count) to null/not null.
+ */
+ public abstract void putNulls(int rowId, int count);
+ public abstract void putNotNulls(int rowId, int count);
+
+ /**
+ * Returns whether the value at rowId is NULL.
+ */
+ public abstract boolean getIsNull(int rowId);
+
+ /**
+ * Sets the value at rowId to `value`.
+ */
+ public abstract void putInt(int rowId, int value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to value.
+ */
+ public abstract void putInts(int rowId, int count, int value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ */
+ public abstract void putInts(int rowId, int count, int[] src, int srcIndex);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * The data in src must be 4-byte little endian ints.
+ */
+ public abstract void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex);
+
+ /**
+ * Returns the integer for rowId.
+ */
+ public abstract int getInt(int rowId);
+
+ /**
+ * Sets the value at rowId to `value`.
+ */
+ public abstract void putDouble(int rowId, double value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to value.
+ */
+ public abstract void putDoubles(int rowId, int count, double value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * src should contain `count` doubles written as ieee format.
+ */
+ public abstract void putDoubles(int rowId, int count, double[] src, int srcIndex);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * The data in src must be ieee formated doubles.
+ */
+ public abstract void putDoubles(int rowId, int count, byte[] src, int srcIndex);
+
+ /**
+ * Returns the double for rowId.
+ */
+ public abstract double getDouble(int rowId);
+
+ /**
+ * Maximum number of rows that can be stored in this column.
+ */
+ protected final int capacity;
+
+ /**
+ * Byte width fo this column.
+ */
+ protected final int width;
+
+ /**
+ * Number of nulls in this column.
+ */
+ protected int numNulls;
+
+ /**
+ * True if there is at least one NULL byte set.
+ */
+ protected boolean anyNullsSet;
--- End diff --
No, I updated the comments. Let me know if it helps.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49376643
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarBatch.java ---
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.util.Iterator;
+
+import org.apache.commons.lang.NotImplementedException;
+
+/**
+ * This class is the in memory representation of rows as they are streamed through operators. It
+ * is designed to maximize CPU efficiency and not storage footprint. Since it is expected that
+ * each operator allocates one of thee objects, the storage footprint on the task is negligible.
+ *
+ * The layout is a columnar with values encoded in their native format. Each RowBatch contains
+ * a horizontal partitioning of the data, split into columns.
+ *
+ * The ColumnarBatch supports either on heap or offheap modes with (mostly) the identical API.
+ *
+ * TODO:
+ * - There are many TODOs for the existing APIs. They should throw a not implemented exception.
+ * - Compaction: The batch and columns should be able to compact based on a selection vector.
+ */
+public final class ColumnarBatch {
+ private static final int DEFAULT_BATCH_SIZE = 4 * 1024;
+
+ private final StructType schema;
+ private final int capacity;
--- End diff --
Yea, I'm working on the Put APIs to grow. I think there's a case when you know (decoding form parquet, from cache) and the other case. We should support both.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49399155
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchBenchmark.scala ---
@@ -0,0 +1,272 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.nio.ByteBuffer
+
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow
+import org.apache.spark.sql.execution.vectorized.ColumnVector
+import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.unsafe.Platform
+import org.apache.spark.util.Benchmark
+
+/**
+ * Benchmark to low level memory access using different ways to manage buffers.
+ */
+object ColumnarBatchBenchmark {
+
+ // This benchmark reads and writes an array of ints.
+ // TODO: there is a big (2x) penalty for a random access API for off heap.
+ // Note: carefully if modifying this code. It's hard to reason about the JIT.
+ def intAccess(iters: Long): Unit = {
+ val count = 8 * 1000
+
+ // Accessing a java array.
+ val javaArray = { i: Int =>
+ val data = new Array[Int](count)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ data(i) = i
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += data(i)
+ i += 1
+ }
+ }
+ }
+
+ // Accessing ByteBuffers
+ val byteBufferUnsafe = { i: Int =>
+ val data = ByteBuffer.allocate(count * 4)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ Platform.putInt(data.array(), Platform.BYTE_ARRAY_OFFSET + i * 4, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += Platform.getInt(data.array(), Platform.BYTE_ARRAY_OFFSET + i * 4)
+ i += 1
+ }
+ }
+ }
+
+ // Accessing offheap byte buffers
+ val directByteBuffer = { i: Int =>
+ val data = ByteBuffer.allocateDirect(count * 4).asIntBuffer()
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ data.put(i)
+ i += 1
+ }
+ data.rewind()
+ i = 0
+ while (i < count) {
+ sum += data.get()
+ i += 1
+ }
+ data.rewind()
+ }
+ }
+
+ // Accessing ByteBuffer using the typed APIs
+ val byteBufferApi = { i: Int =>
+ val data = ByteBuffer.allocate(count * 4)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ data.putInt(i)
+ i += 1
+ }
+ data.rewind()
+ i = 0
+ while (i < count) {
+ sum += data.getInt()
+ i += 1
+ }
+ data.rewind()
+ }
+ }
+
+ // Using unsafe memory
+ val unsafeBuffer = { i: Int =>
+ val data: Long = Platform.allocateMemory(count * 4)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var ptr = data
+ var i = 0
+ while (i < count) {
+ Platform.putInt(null, ptr, i)
+ ptr += 4
+ i += 1
+ }
+ ptr = data
+ i = 0
+ while (i < count) {
+ sum += Platform.getInt(null, ptr)
+ ptr += 4
+ i += 1
+ }
+ }
+ }
+
+ // Access through the column API with on heap memory
+ val columnOnHeap = { i: Int =>
+ val col = ColumnVector.allocate(count, IntegerType, false)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ col.putInt(i, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += col.getInt(i)
+ i += 1
+ }
+ }
+ col.close
+ }
+
+ // Access through the column API with off heap memory
+ def columnOffHeap = { i: Int => {
+ val col = ColumnVector.allocate(count, IntegerType, true)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ col.putInt(i, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += col.getInt(i)
+ i += 1
+ }
+ }
+ col.close
+ }}
+
+ // Access by directly getting the buffer backing the column.
+ val columnOffheapDirect = { i: Int =>
+ val col = ColumnVector.allocate(count, IntegerType, true)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var addr = col.valuesNativeAddress()
+ var i = 0
+ while (i < count) {
+ Platform.putInt(null, addr, i)
+ addr += 4
+ i += 1
+ }
+ i = 0
+ addr = col.valuesNativeAddress()
+ while (i < count) {
+ sum += Platform.getInt(null, addr)
+ addr += 4
+ i += 1
+ }
+ }
+ col.close
+ }
+
+ // Access by going through a batch of unsafe rows.
+ val unsafeRowOnheap = { i: Int =>
+ val buffer = new Array[Byte](count * 16)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ val row = new UnsafeRow(1)
+ var i = 0
+ while (i < count) {
+ row.pointTo(buffer, Platform.BYTE_ARRAY_OFFSET + i * 16, 16)
+ row.setInt(0, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ row.pointTo(buffer, Platform.BYTE_ARRAY_OFFSET + i * 16, 16)
+ sum += row.getInt(0)
+ i += 1
+ }
+ }
+ }
+
+ // Access by going through a batch of unsafe rows.
+ val unsafeRowOffheap = { i: Int =>
+ val buffer = Platform.allocateMemory(count * 16)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ val row = new UnsafeRow(1)
+ var i = 0
+ while (i < count) {
+ row.pointTo(null, buffer + i * 16, 16)
+ row.setInt(0, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ row.pointTo(null, buffer + i * 16, 16)
+ sum += row.getInt(0)
+ i += 1
+ }
+ }
+ Platform.freeMemory(buffer)
+ }
+
+ /*
+ Intel(R) Core(TM) i7-4870HQ CPU @ 2.50GHz
+ Int Read/Write: Avg Time(ms) Avg Rate(M/s) Relative Rate
+ -------------------------------------------------------------------------
+ Java Array 248.8 1317.04 1.00 X
+ ByteBuffer Unsafe 435.6 752.25 0.57 X
+ ByteBuffer API 1752.0 187.03 0.14 X
+ DirectByteBuffer 595.4 550.35 0.42 X
+ Unsafe Buffer 235.2 1393.20 1.06 X
+ Column(on heap) 189.8 1726.45 1.31 X
--- End diff --
I agree it doesn't make sense. I think we're really susceptible to jvm jitting behavior.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49045800
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchBenchmark.scala ---
@@ -0,0 +1,272 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.nio.ByteBuffer
+
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow
+import org.apache.spark.sql.execution.vectorized.ColumnVector
+import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.unsafe.Platform
+import org.apache.spark.util.Benchmark
+
+/**
+ * Benchmark to low level memory access using different ways to manage buffers.
+ */
+object ColumnarBatchBenchmark {
+
+ // This benchmark reads and writes an array of ints.
+ // TODO: there is a big (2x) penalty for a random access API for off heap.
+ // Note: carefully if modifying this code. It's hard to reason about the JIT.
+ def intAccess(iters: Long): Unit = {
+ val count = 8 * 1000
+
+ // Accessing a java array.
+ val javaArray = { i: Int =>
+ val data = new Array[Int](count)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ data(i) = i
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += data(i)
+ i += 1
+ }
+ }
+ }
+
+ // Accessing ByteBuffers
+ val byteBufferUnsafe = { i: Int =>
+ val data = ByteBuffer.allocate(count * 4)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ Platform.putInt(data.array(), Platform.BYTE_ARRAY_OFFSET + i * 4, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += Platform.getInt(data.array(), Platform.BYTE_ARRAY_OFFSET + i * 4)
+ i += 1
+ }
+ }
+ }
+
+ // Accessing offheap byte buffers
+ val directByteBuffer = { i: Int =>
+ val data = ByteBuffer.allocateDirect(count * 4).asIntBuffer()
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ data.put(i)
+ i += 1
+ }
+ data.rewind()
+ i = 0
+ while (i < count) {
+ sum += data.get()
+ i += 1
+ }
+ data.rewind()
+ }
+ }
+
+ // Accessing ByteBuffer using the typed APIs
+ val byteBufferApi = { i: Int =>
+ val data = ByteBuffer.allocate(count * 4)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ data.putInt(i)
+ i += 1
+ }
+ data.rewind()
+ i = 0
+ while (i < count) {
+ sum += data.getInt()
+ i += 1
+ }
+ data.rewind()
+ }
+ }
+
+ // Using unsafe memory
+ val unsafeBuffer = { i: Int =>
+ val data: Long = Platform.allocateMemory(count * 4)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var ptr = data
+ var i = 0
+ while (i < count) {
+ Platform.putInt(null, ptr, i)
+ ptr += 4
+ i += 1
+ }
+ ptr = data
+ i = 0
+ while (i < count) {
+ sum += Platform.getInt(null, ptr)
+ ptr += 4
+ i += 1
+ }
+ }
+ }
+
+ // Access through the column API with on heap memory
+ val columnOnHeap = { i: Int =>
+ val col = ColumnVector.allocate(count, IntegerType, false)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ col.putInt(i, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += col.getInt(i)
+ i += 1
+ }
+ }
+ col.close
+ }
+
+ // Access through the column API with off heap memory
+ def columnOffHeap = { i: Int => {
+ val col = ColumnVector.allocate(count, IntegerType, true)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ col.putInt(i, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += col.getInt(i)
+ i += 1
+ }
+ }
+ col.close
+ }}
+
+ // Access by directly getting the buffer backing the column.
+ val columnOffheapDirect = { i: Int =>
+ val col = ColumnVector.allocate(count, IntegerType, true)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var addr = col.valuesNativeAddress()
+ var i = 0
+ while (i < count) {
+ Platform.putInt(null, addr, i)
+ addr += 4
+ i += 1
+ }
+ i = 0
+ addr = col.valuesNativeAddress()
+ while (i < count) {
+ sum += Platform.getInt(null, addr)
+ addr += 4
+ i += 1
+ }
+ }
+ col.close
+ }
+
+ // Access by going through a batch of unsafe rows.
+ val unsafeRowOnheap = { i: Int =>
+ val buffer = new Array[Byte](count * 16)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ val row = new UnsafeRow(1)
--- End diff --
Should this be out of the loop?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49375889
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java ---
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+
+/**
+ * This class represents a column of values and provides the main APIs to access the data
+ * values. It supports all the types and contains get/put APIs as well as their batched versions.
+ * The batched versions are preferable whenever possible.
+ *
+ * Most of the APIs take the rowId as a parameter. This is the local 0-based row id for values
+ * in the current RowBatch.
+ *
+ * A ColumnVector should be considered immutable once originally created. In other words, it is not
+ * valid to call put APIs after reads until reset() is called.
+ */
+public abstract class ColumnVector {
+ /**
+ * Allocates a column with each element of size `width` either on or off heap.
+ */
+ public static ColumnVector allocate(int capacity, DataType type, boolean offHeap) {
+ if (offHeap) {
+ return new OffHeapColumnVector(capacity, type);
+ } else {
+ return new OnHeapColumnVector(capacity, type);
+ }
+ }
+
+ public final DataType dataType() { return type; }
+
+ /**
+ * Resets this column for writing. The currently stored values are no longer accessible.
+ */
+ public void reset() {
+ numNulls = 0;
+ if (anyNullsSet) {
+ putNotNulls(0, capacity);
+ anyNullsSet = false;
+ }
+ }
+
+ /**
+ * Cleans up memory for this column. The column is not usable after this.
+ * TODO: this should probably have ref-counted semantics.
+ */
+ public abstract void close();
+
+ /**
+ * Returns the number of nulls in this column.
+ */
+ public final int numNulls() { return numNulls; }
+
+ /**
+ * Returns true if any of the nulls indicator are set for this column. This can be used
+ * as an optimization to prevent setting nulls.
+ */
+ public final boolean anyNullsSet() { return anyNullsSet; }
+
+ /**
+ * Returns the off heap ptr for the arrays backing the NULLs and values buffer. Only valid
+ * to call for off heap columns.
+ */
+ public abstract long nullsNativeAddress();
+ public abstract long valuesNativeAddress();
+
+ /**
+ * Sets the value at rowId to null/not null.
+ */
+ public abstract void putNotNull(int rowId);
+ public abstract void putNull(int rowId);
+
+ /**
+ * Sets the values from [rowId, rowId + count) to null/not null.
+ */
+ public abstract void putNulls(int rowId, int count);
+ public abstract void putNotNulls(int rowId, int count);
+
+ /**
+ * Returns whether the value at rowId is NULL.
+ */
+ public abstract boolean getIsNull(int rowId);
+
+ /**
+ * Sets the value at rowId to `value`.
+ */
+ public abstract void putInt(int rowId, int value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to value.
+ */
+ public abstract void putInts(int rowId, int count, int value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ */
+ public abstract void putInts(int rowId, int count, int[] src, int srcIndex);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * The data in src must be 4-byte little endian ints.
+ */
+ public abstract void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex);
+
+ /**
+ * Returns the integer for rowId.
+ */
+ public abstract int getInt(int rowId);
+
+ /**
+ * Sets the value at rowId to `value`.
+ */
+ public abstract void putDouble(int rowId, double value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to value.
+ */
+ public abstract void putDoubles(int rowId, int count, double value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * src should contain `count` doubles written as ieee format.
+ */
+ public abstract void putDoubles(int rowId, int count, double[] src, int srcIndex);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * The data in src must be ieee formated doubles.
+ */
+ public abstract void putDoubles(int rowId, int count, byte[] src, int srcIndex);
+
+ /**
+ * Returns the double for rowId.
+ */
+ public abstract double getDouble(int rowId);
+
+ /**
+ * Maximum number of rows that can be stored in this column.
+ */
+ protected final int capacity;
+
+ /**
+ * Byte width fo this column.
+ */
+ protected final int width;
+
+ /**
+ * Number of nulls in this column.
+ */
+ protected int numNulls;
+
+ /**
+ * True if there is at least one NULL byte set.
+ */
+ protected boolean anyNullsSet;
+
+ /**
+ * Data type for this column.
+ */
+ protected final DataType type;
+
+ protected ColumnVector(int capacity, DataType type) {
+ this.capacity = capacity;
+ this.type = type;
+ this.width = type.defaultSize();
--- End diff --
I don't use this right now. Removed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49400036
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java ---
@@ -0,0 +1,162 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * Column data backed using offheap memory.
+ */
+public final class OffHeapColumnVector extends ColumnVector {
+ // The data stored in these two allocations need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+ private long nulls;
+ private long data;
+
+ protected OffHeapColumnVector(int capacity, DataType type) {
+ super(capacity, type);
+ this.nulls = Platform.allocateMemory(capacity);
+ if (type instanceof IntegerType) {
+ this.data = Platform.allocateMemory(capacity * 4);
+ } else if (type instanceof DoubleType) {
+ this.data = Platform.allocateMemory(capacity * 8);
+ } else {
+ throw new RuntimeException("Unhandled " + type);
+ }
+ reset();
+ }
+
+ @Override
+ public final long valuesNativeAddress() {
+ return data;
+ }
+
+ @Override
+ public long nullsNativeAddress() {
+ return nulls;
+ }
+
+ @Override
+ public final void close() {
+ Platform.freeMemory(nulls);
+ Platform.freeMemory(data);
+ nulls = 0;
+ data = 0;
+ }
+
+ //
+ // APIs dealing with nulls
+ //
+
+ @Override
+ public final void putNotNull(int rowId) {
+ Platform.putByte(null, nulls + rowId, (byte) 0);
+ }
+
+ @Override
+ public final void putNull(int rowId) {
+ Platform.putByte(null, nulls + rowId, (byte) 1);
+ ++numNulls;
+ anyNullsSet = true;
+ }
+
+ @Override
+ public final void putNulls(int rowId, int count) {
+ long offset = nulls + rowId;
+ for (int i = 0; i < count; ++i, ++offset) {
+ Platform.putByte(null, offset, (byte) 1);
+ }
+ anyNullsSet = true;
+ numNulls += count;
+ }
+
+ @Override
+ public final void putNotNulls(int rowId, int count) {
+ long offset = nulls + rowId;
+ for (int i = 0; i < count; ++i, ++offset) {
+ Platform.putByte(null, offset, (byte) 0);
+ }
+ }
+
+ @Override
+ public final boolean getIsNull(int rowId) {
+ return Platform.getByte(null, nulls + rowId) == 1;
+ }
+
+ //
+ // APIs dealing with ints
+ //
+
+ @Override
+ public final void putInt(int rowId, int value) {
+ Platform.putInt(null, data + 4 * rowId, value);
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int value) {
+ long offset = data + 4 * rowId;
+ for (int i = 0; i < count; ++i, offset += 4) {
+ Platform.putInt(null, offset, value);
+ }
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int[] src, int srcIndex) {
+ Platform.copyMemory(src, Platform.INT_ARRAY_OFFSET + srcIndex * 4,
+ null, data + 4 * rowId, count * 4);
+ }
+
+ @Override
+ public final void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex) {
--- End diff --
Done. Added a check and it will throw on big endian
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49399104
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchBenchmark.scala ---
@@ -0,0 +1,272 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.nio.ByteBuffer
+
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow
+import org.apache.spark.sql.execution.vectorized.ColumnVector
+import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.unsafe.Platform
+import org.apache.spark.util.Benchmark
+
+/**
+ * Benchmark to low level memory access using different ways to manage buffers.
+ */
+object ColumnarBatchBenchmark {
+
+ // This benchmark reads and writes an array of ints.
+ // TODO: there is a big (2x) penalty for a random access API for off heap.
+ // Note: carefully if modifying this code. It's hard to reason about the JIT.
+ def intAccess(iters: Long): Unit = {
+ val count = 8 * 1000
+
+ // Accessing a java array.
+ val javaArray = { i: Int =>
+ val data = new Array[Int](count)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ data(i) = i
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += data(i)
+ i += 1
+ }
+ }
+ }
+
+ // Accessing ByteBuffers
+ val byteBufferUnsafe = { i: Int =>
+ val data = ByteBuffer.allocate(count * 4)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ Platform.putInt(data.array(), Platform.BYTE_ARRAY_OFFSET + i * 4, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += Platform.getInt(data.array(), Platform.BYTE_ARRAY_OFFSET + i * 4)
+ i += 1
+ }
+ }
+ }
+
+ // Accessing offheap byte buffers
+ val directByteBuffer = { i: Int =>
+ val data = ByteBuffer.allocateDirect(count * 4).asIntBuffer()
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ data.put(i)
+ i += 1
+ }
+ data.rewind()
+ i = 0
+ while (i < count) {
+ sum += data.get()
+ i += 1
+ }
+ data.rewind()
+ }
+ }
+
+ // Accessing ByteBuffer using the typed APIs
+ val byteBufferApi = { i: Int =>
+ val data = ByteBuffer.allocate(count * 4)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ data.putInt(i)
+ i += 1
+ }
+ data.rewind()
+ i = 0
+ while (i < count) {
+ sum += data.getInt()
+ i += 1
+ }
+ data.rewind()
+ }
+ }
+
+ // Using unsafe memory
+ val unsafeBuffer = { i: Int =>
+ val data: Long = Platform.allocateMemory(count * 4)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var ptr = data
+ var i = 0
+ while (i < count) {
+ Platform.putInt(null, ptr, i)
+ ptr += 4
+ i += 1
+ }
+ ptr = data
+ i = 0
+ while (i < count) {
+ sum += Platform.getInt(null, ptr)
+ ptr += 4
+ i += 1
+ }
+ }
+ }
+
+ // Access through the column API with on heap memory
+ val columnOnHeap = { i: Int =>
+ val col = ColumnVector.allocate(count, IntegerType, false)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ col.putInt(i, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += col.getInt(i)
+ i += 1
+ }
+ }
+ col.close
+ }
+
+ // Access through the column API with off heap memory
+ def columnOffHeap = { i: Int => {
+ val col = ColumnVector.allocate(count, IntegerType, true)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ col.putInt(i, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += col.getInt(i)
+ i += 1
+ }
+ }
+ col.close
+ }}
+
+ // Access by directly getting the buffer backing the column.
+ val columnOffheapDirect = { i: Int =>
+ val col = ColumnVector.allocate(count, IntegerType, true)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var addr = col.valuesNativeAddress()
+ var i = 0
+ while (i < count) {
+ Platform.putInt(null, addr, i)
+ addr += 4
+ i += 1
+ }
+ i = 0
+ addr = col.valuesNativeAddress()
+ while (i < count) {
+ sum += Platform.getInt(null, addr)
+ addr += 4
+ i += 1
+ }
+ }
+ col.close
+ }
+
+ // Access by going through a batch of unsafe rows.
+ val unsafeRowOnheap = { i: Int =>
+ val buffer = new Array[Byte](count * 16)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ val row = new UnsafeRow(1)
--- End diff --
It's already not in the inner loop. Count is big enough.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by kiszk <gi...@git.apache.org>.
Github user kiszk commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49050418
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java ---
@@ -0,0 +1,162 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * Column data backed using offheap memory.
+ */
+public final class OffHeapColumnVector extends ColumnVector {
+ // The data stored in these two allocations need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+ private long nulls;
+ private long data;
+
+ protected OffHeapColumnVector(int capacity, DataType type) {
+ super(capacity, type);
+ this.nulls = Platform.allocateMemory(capacity);
--- End diff --
Is is better to first get a area for this.data and to next get a area for this.nulls for this.nulls from the combined allocated memory? It expects to this.data is aligned with a cache line boundary.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by yjshen <gi...@git.apache.org>.
Github user yjshen commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-169646900
Is this PR the first step to enable vectorized execution in Spark SQL or just for efficient serialization to use external components?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12785][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-204503727
This is not an architectural limitation. We have just not implemented support for it and there are only a few methods that would need to be implemented to support big endian. It would be great if someone in the community working on big endian hardware could do this.
We don't rely on the byte ordering of the platform. The function you are referring to, putIntLittleEndian, is converting the input, which is little endian, to whatever the machine's endianness is. It doesn't specify what the host endian has to be. This is used in cases where the data is encoded on disk in a canonical binary format. On a big endian host, this would have to byte swap but I think that's inevitable as the on disk data had to pick an endianness (this is the code that's not implemented right now).
If you find places where spark requires a particular endianness, i'd consider that a bug.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by yjshen <gi...@git.apache.org>.
Github user yjshen commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49151061
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java ---
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+
+/**
+ * This class represents a column of values and provides the main APIs to access the data
+ * values. It supports all the types and contains get/put APIs as well as their batched versions.
+ * The batched versions are preferable whenever possible.
+ *
+ * Most of the APIs take the rowId as a parameter. This is the local 0-based row id for values
+ * in the current RowBatch.
+ *
+ * A ColumnVector should be considered immutable once originally created. In other words, it is not
+ * valid to call put APIs after reads until reset() is called.
+ */
+public abstract class ColumnVector {
+ /**
+ * Allocates a column with each element of size `width` either on or off heap.
+ */
+ public static ColumnVector allocate(int capacity, DataType type, boolean offHeap) {
+ if (offHeap) {
+ return new OffHeapColumnVector(capacity, type);
+ } else {
+ return new OnHeapColumnVector(capacity, type);
+ }
+ }
+
+ public final DataType dataType() { return type; }
+
+ /**
+ * Resets this column for writing. The currently stored values are no longer accessible.
+ */
+ public void reset() {
+ numNulls = 0;
+ if (anyNullsSet) {
+ putNotNulls(0, capacity);
+ anyNullsSet = false;
+ }
+ }
+
+ /**
+ * Cleans up memory for this column. The column is not usable after this.
+ * TODO: this should probably have ref-counted semantics.
+ */
+ public abstract void close();
+
+ /**
+ * Returns the number of nulls in this column.
+ */
+ public final int numNulls() { return numNulls; }
+
+ /**
+ * Returns true if any of the nulls indicator are set for this column. This can be used
+ * as an optimization to prevent setting nulls.
+ */
+ public final boolean anyNullsSet() { return anyNullsSet; }
+
+ /**
+ * Returns the off heap ptr for the arrays backing the NULLs and values buffer. Only valid
+ * to call for off heap columns.
+ */
+ public abstract long nullsNativeAddress();
+ public abstract long valuesNativeAddress();
+
+ /**
+ * Sets the value at rowId to null/not null.
+ */
+ public abstract void putNotNull(int rowId);
+ public abstract void putNull(int rowId);
+
+ /**
+ * Sets the values from [rowId, rowId + count) to null/not null.
+ */
+ public abstract void putNulls(int rowId, int count);
+ public abstract void putNotNulls(int rowId, int count);
+
+ /**
+ * Returns whether the value at rowId is NULL.
+ */
+ public abstract boolean getIsNull(int rowId);
+
+ /**
+ * Sets the value at rowId to `value`.
+ */
+ public abstract void putInt(int rowId, int value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to value.
+ */
+ public abstract void putInts(int rowId, int count, int value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ */
+ public abstract void putInts(int rowId, int count, int[] src, int srcIndex);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * The data in src must be 4-byte little endian ints.
+ */
+ public abstract void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex);
+
+ /**
+ * Returns the integer for rowId.
+ */
+ public abstract int getInt(int rowId);
+
+ /**
+ * Sets the value at rowId to `value`.
+ */
+ public abstract void putDouble(int rowId, double value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to value.
+ */
+ public abstract void putDoubles(int rowId, int count, double value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * src should contain `count` doubles written as ieee format.
+ */
+ public abstract void putDoubles(int rowId, int count, double[] src, int srcIndex);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * The data in src must be ieee formated doubles.
+ */
+ public abstract void putDoubles(int rowId, int count, byte[] src, int srcIndex);
+
+ /**
+ * Returns the double for rowId.
+ */
+ public abstract double getDouble(int rowId);
+
+ /**
+ * Maximum number of rows that can be stored in this column.
+ */
+ protected final int capacity;
+
+ /**
+ * Byte width fo this column.
+ */
+ protected final int width;
+
+ /**
+ * Number of nulls in this column.
+ */
+ protected int numNulls;
+
+ /**
+ * True if there is at least one NULL byte set.
+ */
+ protected boolean anyNullsSet;
+
+ /**
+ * Data type for this column.
+ */
+ protected final DataType type;
+
+ protected ColumnVector(int capacity, DataType type) {
+ this.capacity = capacity;
+ this.type = type;
+ this.width = type.defaultSize();
+ }
+}
--- End diff --
Maybe we could add an `isRepeating` Flag here for special ColumnVectors, e.g. partition key column, literal column, and flatten them if needed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49045594
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java ---
@@ -0,0 +1,165 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * A column backed by an in memory JVM array. This stores the NULLs as a byte per value
+ * and a java array for the values.
+ */
+public final class OnHeapColumnVector extends ColumnVector {
+ // The data stored in these arrays need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+
+ // This is faster than a boolean array and we optimize this over memory footprint.
--- End diff --
I think a bitset is usually faster than a byte array
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12785][SQL] Add ColumnarBatch, an in me...
Posted by robbinspg <gi...@git.apache.org>.
Github user robbinspg commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-209832195
@nongli I'm just about there with a solution for Big Endian platforms and will be using https://issues.apache.org/jira/browse/SPARK-14151 for the changes.
I have one question:
It is clear from the tests using Parquet that the byte array passed into putIntsLittleEndian is in little endian order. It is also the case that the byte array passed in to the putFloats and putDoubles has the values in little endian. Reversing the floats/doubles enables all the tests to pass.
In OffHeapColumnVector putDoubles(int rowId, int count, byte[] src, int srcIndex) if I assume input is LE then the org.apache.spark.sql.execution.vectorized.ColumnarBatchSuite.Double APIs test fails. This is because it is passing in a byte array of doubles that are in platform endian order (created with Platform.putDouble).
My question is: are the byte arrays always in little endian? This seems to be true for the Parquet sources?? If so then I can modify the testcase 'org.apache.spark.sql.execution.vectorized.ColumnarBatchSuite.Double APIs' to force the test data into LE.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12785][SQL] Add ColumnarBatch, an in me...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/10628
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49376182
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarBatch.java ---
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.util.Iterator;
+
+import org.apache.commons.lang.NotImplementedException;
+
+/**
+ * This class is the in memory representation of rows as they are streamed through operators. It
+ * is designed to maximize CPU efficiency and not storage footprint. Since it is expected that
+ * each operator allocates one of thee objects, the storage footprint on the task is negligible.
+ *
+ * The layout is a columnar with values encoded in their native format. Each RowBatch contains
+ * a horizontal partitioning of the data, split into columns.
+ *
+ * The ColumnarBatch supports either on heap or offheap modes with (mostly) the identical API.
+ *
+ * TODO:
+ * - There are many TODOs for the existing APIs. They should throw a not implemented exception.
+ * - Compaction: The batch and columns should be able to compact based on a selection vector.
+ */
+public final class ColumnarBatch {
+ private static final int DEFAULT_BATCH_SIZE = 4 * 1024;
+
+ private final StructType schema;
+ private final int capacity;
+ private int numRows;
+ private final ColumnVector[] columns;
+
+ // True if the row is filtered.
+ private final boolean[] filteredRows;
+
+ // Total number of rows that have been filtered.
+ private int numRowsFiltered = 0;
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap) {
+ return new ColumnarBatch(schema, DEFAULT_BATCH_SIZE, offHeap);
+ }
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap, int maxRows) {
+ return new ColumnarBatch(schema, maxRows, offHeap);
+ }
+
+ /**
+ * Called to close all the columns in this batch. It is not valid to access the data after
+ * calling this. This must be called at the end to clean up memory allcoations.
+ */
+ public void close() {
+ for (ColumnVector c: columns) {
+ c.close();
+ }
+ }
+
+ /**
+ * Adapter class to interop with existing components that expect internal row. A lot of
+ * performance is lost with this translation.
+ */
+ public final class Row extends InternalRow {
+ private int rowId;
+
+ /**
+ * Marks this row as being filtered out. This means a subsequent iteration over the rows
+ * in this batch will not include this row.
+ */
+ public final void markFiltered() {
+ ColumnarBatch.this.markFiltered(rowId);
+ }
+
+ @Override
+ public final int numFields() {
+ return ColumnarBatch.this.numCols();
+ }
+
+ @Override
+ public final InternalRow copy() {
+ throw new NotImplementedException();
--- End diff --
Yea, I do this in a subsequent patch.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-169552399
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49044546
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarBatch.java ---
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.util.Iterator;
+
+import org.apache.commons.lang.NotImplementedException;
+
+/**
+ * This class is the in memory representation of rows as they are streamed through operators. It
+ * is designed to maximize CPU efficiency and not storage footprint. Since it is expected that
+ * each operator allocates one of thee objects, the storage footprint on the task is negligible.
+ *
+ * The layout is a columnar with values encoded in their native format. Each RowBatch contains
+ * a horizontal partitioning of the data, split into columns.
+ *
+ * The ColumnarBatch supports either on heap or offheap modes with (mostly) the identical API.
+ *
+ * TODO:
+ * - There are many TODOs for the existing APIs. They should throw a not implemented exception.
+ * - Compaction: The batch and columns should be able to compact based on a selection vector.
+ */
+public final class ColumnarBatch {
+ private static final int DEFAULT_BATCH_SIZE = 4 * 1024;
+
+ private final StructType schema;
+ private final int capacity;
+ private int numRows;
+ private final ColumnVector[] columns;
+
+ // True if the row is filtered.
+ private final boolean[] filteredRows;
+
+ // Total number of rows that have been filtered.
+ private int numRowsFiltered = 0;
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap) {
+ return new ColumnarBatch(schema, DEFAULT_BATCH_SIZE, offHeap);
+ }
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap, int maxRows) {
+ return new ColumnarBatch(schema, maxRows, offHeap);
+ }
+
+ /**
+ * Called to close all the columns in this batch. It is not valid to access the data after
+ * calling this. This must be called at the end to clean up memory allcoations.
+ */
+ public void close() {
+ for (ColumnVector c: columns) {
+ c.close();
+ }
+ }
+
+ /**
+ * Adapter class to interop with existing components that expect internal row. A lot of
+ * performance is lost with this translation.
+ */
+ public final class Row extends InternalRow {
+ private int rowId;
+
+ /**
+ * Marks this row as being filtered out. This means a subsequent iteration over the rows
+ * in this batch will not include this row.
+ */
+ public final void markFiltered() {
+ ColumnarBatch.this.markFiltered(rowId);
+ }
+
+ @Override
+ public final int numFields() {
+ return ColumnarBatch.this.numCols();
+ }
+
+ @Override
+ public final InternalRow copy() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final boolean anyNull() {
+ return false;
--- End diff --
Should this be NotImplementedException?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49045624
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java ---
@@ -0,0 +1,165 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * A column backed by an in memory JVM array. This stores the NULLs as a byte per value
+ * and a java array for the values.
+ */
+public final class OnHeapColumnVector extends ColumnVector {
+ // The data stored in these arrays need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+
+ // This is faster than a boolean array and we optimize this over memory footprint.
+ private byte[] nulls;
+
+ // Array for each type. Only 1 is populated for any type.
+ private int[] intData;
+ private double[] doubleData;
+
+ protected OnHeapColumnVector(int capacity, DataType type) {
+ super(capacity, type);
+ if (type instanceof IntegerType) {
+ this.intData = new int[capacity];
+ } else if (type instanceof DoubleType) {
+ this.doubleData = new double[capacity];
+ } else {
+ throw new RuntimeException("Unhandled " + type);
+ }
+ this.nulls = new byte[capacity];
+ reset();
+ }
+
+ @Override
+ public final long valuesNativeAddress() {
+ throw new RuntimeException("Cannot get native address for on heap column");
+ }
+ @Override
+ public final long nullsNativeAddress() {
+ throw new RuntimeException("Cannot get native address for on heap column");
+ }
+
+ @Override
+ public final void close() {
+ nulls = null;
+ intData = null;
+ doubleData = null;
+ }
+
+
+ //
+ // APIs dealing with nulls
+ //
+
+ @Override
+ public final void putNotNull(int rowId) {
+ nulls[rowId] = (byte)0;
+ }
+
+ @Override
+ public final void putNull(int rowId) {
+ nulls[rowId] = (byte)1;
+ ++numNulls;
+ anyNullsSet = true;
+ }
+
+ @Override
+ public final void putNulls(int rowId, int count) {
+ for (int i = 0; i < count; ++i) {
+ nulls[rowId + i] = (byte)1;
+ }
+ anyNullsSet = true;
+ numNulls += count;
+ }
+
+ @Override
+ public final void putNotNulls(int rowId, int count) {
+ for (int i = 0; i < count; ++i) {
+ nulls[rowId + i] = (byte)0;
+ }
+ }
+
+ @Override
+ public final boolean getIsNull(int rowId) {
+ return nulls[rowId] == 1;
+ }
+
+ //
+ // APIs dealing with Ints
+ //
+
+ @Override
+ public final void putInt(int rowId, int value) {
+ intData[rowId] = value;
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int value) {
+ for (int i = 0; i < count; ++i) {
+ intData[i + rowId] = value;
+ }
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int[] src, int srcIndex) {
+ System.arraycopy(src, srcIndex, intData, rowId, count);
+ }
+
+ @Override
+ public final void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex) {
+ for (int i = 0; i < count; ++i) {
+ // TODO: is this the fastest way?
--- End diff --
We could still use Platform.copyMemory() here
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49045388
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java ---
@@ -0,0 +1,162 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * Column data backed using offheap memory.
+ */
+public final class OffHeapColumnVector extends ColumnVector {
+ // The data stored in these two allocations need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+ private long nulls;
+ private long data;
+
+ protected OffHeapColumnVector(int capacity, DataType type) {
+ super(capacity, type);
+ this.nulls = Platform.allocateMemory(capacity);
--- End diff --
minor: This can be allocated together with data (at the end).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49376808
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java ---
@@ -0,0 +1,162 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * Column data backed using offheap memory.
+ */
+public final class OffHeapColumnVector extends ColumnVector {
+ // The data stored in these two allocations need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+ private long nulls;
+ private long data;
+
+ protected OffHeapColumnVector(int capacity, DataType type) {
+ super(capacity, type);
+ this.nulls = Platform.allocateMemory(capacity);
+ if (type instanceof IntegerType) {
+ this.data = Platform.allocateMemory(capacity * 4);
+ } else if (type instanceof DoubleType) {
+ this.data = Platform.allocateMemory(capacity * 8);
+ } else {
+ throw new RuntimeException("Unhandled " + type);
+ }
+ reset();
+ }
+
+ @Override
+ public final long valuesNativeAddress() {
+ return data;
+ }
+
+ @Override
+ public long nullsNativeAddress() {
+ return nulls;
+ }
+
+ @Override
+ public final void close() {
+ Platform.freeMemory(nulls);
+ Platform.freeMemory(data);
+ nulls = 0;
+ data = 0;
+ }
+
+ //
+ // APIs dealing with nulls
+ //
+
+ @Override
+ public final void putNotNull(int rowId) {
+ Platform.putByte(null, nulls + rowId, (byte) 0);
+ }
+
+ @Override
+ public final void putNull(int rowId) {
+ Platform.putByte(null, nulls + rowId, (byte) 1);
+ ++numNulls;
+ anyNullsSet = true;
+ }
+
+ @Override
+ public final void putNulls(int rowId, int count) {
+ long offset = nulls + rowId;
+ for (int i = 0; i < count; ++i, ++offset) {
+ Platform.putByte(null, offset, (byte) 1);
+ }
+ anyNullsSet = true;
+ numNulls += count;
+ }
+
+ @Override
+ public final void putNotNulls(int rowId, int count) {
+ long offset = nulls + rowId;
+ for (int i = 0; i < count; ++i, ++offset) {
+ Platform.putByte(null, offset, (byte) 0);
+ }
+ }
+
+ @Override
+ public final boolean getIsNull(int rowId) {
+ return Platform.getByte(null, nulls + rowId) == 1;
+ }
+
+ //
+ // APIs dealing with ints
+ //
+
+ @Override
+ public final void putInt(int rowId, int value) {
+ Platform.putInt(null, data + 4 * rowId, value);
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int value) {
+ long offset = data + 4 * rowId;
+ for (int i = 0; i < count; ++i, offset += 4) {
+ Platform.putInt(null, offset, value);
+ }
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int[] src, int srcIndex) {
+ Platform.copyMemory(src, Platform.INT_ARRAY_OFFSET + srcIndex * 4,
+ null, data + 4 * rowId, count * 4);
+ }
+
+ @Override
+ public final void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex) {
--- End diff --
I want to be explicit here. It's useful for things like parquet that persist with a specific endianness.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-170762753
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-170744338
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/49194/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49044429
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java ---
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+
+/**
+ * This class represents a column of values and provides the main APIs to access the data
+ * values. It supports all the types and contains get/put APIs as well as their batched versions.
+ * The batched versions are preferable whenever possible.
+ *
+ * Most of the APIs take the rowId as a parameter. This is the local 0-based row id for values
+ * in the current RowBatch.
+ *
+ * A ColumnVector should be considered immutable once originally created. In other words, it is not
+ * valid to call put APIs after reads until reset() is called.
+ */
+public abstract class ColumnVector {
+ /**
+ * Allocates a column with each element of size `width` either on or off heap.
+ */
+ public static ColumnVector allocate(int capacity, DataType type, boolean offHeap) {
+ if (offHeap) {
+ return new OffHeapColumnVector(capacity, type);
+ } else {
+ return new OnHeapColumnVector(capacity, type);
+ }
+ }
+
+ public final DataType dataType() { return type; }
+
+ /**
+ * Resets this column for writing. The currently stored values are no longer accessible.
+ */
+ public void reset() {
+ numNulls = 0;
+ if (anyNullsSet) {
+ putNotNulls(0, capacity);
+ anyNullsSet = false;
+ }
+ }
+
+ /**
+ * Cleans up memory for this column. The column is not usable after this.
+ * TODO: this should probably have ref-counted semantics.
+ */
+ public abstract void close();
+
+ /**
+ * Returns the number of nulls in this column.
+ */
+ public final int numNulls() { return numNulls; }
+
+ /**
+ * Returns true if any of the nulls indicator are set for this column. This can be used
+ * as an optimization to prevent setting nulls.
+ */
+ public final boolean anyNullsSet() { return anyNullsSet; }
+
+ /**
+ * Returns the off heap ptr for the arrays backing the NULLs and values buffer. Only valid
+ * to call for off heap columns.
+ */
+ public abstract long nullsNativeAddress();
+ public abstract long valuesNativeAddress();
+
+ /**
+ * Sets the value at rowId to null/not null.
+ */
+ public abstract void putNotNull(int rowId);
+ public abstract void putNull(int rowId);
+
+ /**
+ * Sets the values from [rowId, rowId + count) to null/not null.
+ */
+ public abstract void putNulls(int rowId, int count);
+ public abstract void putNotNulls(int rowId, int count);
+
+ /**
+ * Returns whether the value at rowId is NULL.
+ */
+ public abstract boolean getIsNull(int rowId);
+
+ /**
+ * Sets the value at rowId to `value`.
+ */
+ public abstract void putInt(int rowId, int value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to value.
+ */
+ public abstract void putInts(int rowId, int count, int value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ */
+ public abstract void putInts(int rowId, int count, int[] src, int srcIndex);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * The data in src must be 4-byte little endian ints.
+ */
+ public abstract void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex);
+
+ /**
+ * Returns the integer for rowId.
+ */
+ public abstract int getInt(int rowId);
+
+ /**
+ * Sets the value at rowId to `value`.
+ */
+ public abstract void putDouble(int rowId, double value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to value.
+ */
+ public abstract void putDoubles(int rowId, int count, double value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * src should contain `count` doubles written as ieee format.
+ */
+ public abstract void putDoubles(int rowId, int count, double[] src, int srcIndex);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * The data in src must be ieee formated doubles.
+ */
+ public abstract void putDoubles(int rowId, int count, byte[] src, int srcIndex);
+
+ /**
+ * Returns the double for rowId.
+ */
+ public abstract double getDouble(int rowId);
+
+ /**
+ * Maximum number of rows that can be stored in this column.
+ */
+ protected final int capacity;
+
+ /**
+ * Byte width fo this column.
+ */
+ protected final int width;
+
+ /**
+ * Number of nulls in this column.
+ */
+ protected int numNulls;
+
+ /**
+ * True if there is at least one NULL byte set.
+ */
+ protected boolean anyNullsSet;
+
+ /**
+ * Data type for this column.
+ */
+ protected final DataType type;
+
+ protected ColumnVector(int capacity, DataType type) {
+ this.capacity = capacity;
+ this.type = type;
+ this.width = type.defaultSize();
--- End diff --
This is only true for native types
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-170745264
**[Test build #49195 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/49195/consoleFull)** for PR 10628 at commit [`5b27f65`](https://github.com/apache/spark/commit/5b27f651db7b4bab17d0a33d472bf47178b52390).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12785][SQL] Add ColumnarBatch, an in me...
Posted by robbinspg <gi...@git.apache.org>.
Github user robbinspg commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-206732654
We are actually seeing this issue in the OnHeap code as the byte array passed in to putIntLittleEndian is in little endian and the code is trying to read that with Platform.getInt, which on big endian platform will return the wrong value as it assumes the bytes are in big endian
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49046086
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/vectorized/ColumnarBatchBenchmark.scala ---
@@ -0,0 +1,272 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources.parquet
+
+import java.nio.ByteBuffer
+
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow
+import org.apache.spark.sql.execution.vectorized.ColumnVector
+import org.apache.spark.sql.types.IntegerType
+import org.apache.spark.unsafe.Platform
+import org.apache.spark.util.Benchmark
+
+/**
+ * Benchmark to low level memory access using different ways to manage buffers.
+ */
+object ColumnarBatchBenchmark {
+
+ // This benchmark reads and writes an array of ints.
+ // TODO: there is a big (2x) penalty for a random access API for off heap.
+ // Note: carefully if modifying this code. It's hard to reason about the JIT.
+ def intAccess(iters: Long): Unit = {
+ val count = 8 * 1000
+
+ // Accessing a java array.
+ val javaArray = { i: Int =>
+ val data = new Array[Int](count)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ data(i) = i
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += data(i)
+ i += 1
+ }
+ }
+ }
+
+ // Accessing ByteBuffers
+ val byteBufferUnsafe = { i: Int =>
+ val data = ByteBuffer.allocate(count * 4)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ Platform.putInt(data.array(), Platform.BYTE_ARRAY_OFFSET + i * 4, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += Platform.getInt(data.array(), Platform.BYTE_ARRAY_OFFSET + i * 4)
+ i += 1
+ }
+ }
+ }
+
+ // Accessing offheap byte buffers
+ val directByteBuffer = { i: Int =>
+ val data = ByteBuffer.allocateDirect(count * 4).asIntBuffer()
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ data.put(i)
+ i += 1
+ }
+ data.rewind()
+ i = 0
+ while (i < count) {
+ sum += data.get()
+ i += 1
+ }
+ data.rewind()
+ }
+ }
+
+ // Accessing ByteBuffer using the typed APIs
+ val byteBufferApi = { i: Int =>
+ val data = ByteBuffer.allocate(count * 4)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ data.putInt(i)
+ i += 1
+ }
+ data.rewind()
+ i = 0
+ while (i < count) {
+ sum += data.getInt()
+ i += 1
+ }
+ data.rewind()
+ }
+ }
+
+ // Using unsafe memory
+ val unsafeBuffer = { i: Int =>
+ val data: Long = Platform.allocateMemory(count * 4)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var ptr = data
+ var i = 0
+ while (i < count) {
+ Platform.putInt(null, ptr, i)
+ ptr += 4
+ i += 1
+ }
+ ptr = data
+ i = 0
+ while (i < count) {
+ sum += Platform.getInt(null, ptr)
+ ptr += 4
+ i += 1
+ }
+ }
+ }
+
+ // Access through the column API with on heap memory
+ val columnOnHeap = { i: Int =>
+ val col = ColumnVector.allocate(count, IntegerType, false)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ col.putInt(i, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += col.getInt(i)
+ i += 1
+ }
+ }
+ col.close
+ }
+
+ // Access through the column API with off heap memory
+ def columnOffHeap = { i: Int => {
+ val col = ColumnVector.allocate(count, IntegerType, true)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var i = 0
+ while (i < count) {
+ col.putInt(i, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ sum += col.getInt(i)
+ i += 1
+ }
+ }
+ col.close
+ }}
+
+ // Access by directly getting the buffer backing the column.
+ val columnOffheapDirect = { i: Int =>
+ val col = ColumnVector.allocate(count, IntegerType, true)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ var addr = col.valuesNativeAddress()
+ var i = 0
+ while (i < count) {
+ Platform.putInt(null, addr, i)
+ addr += 4
+ i += 1
+ }
+ i = 0
+ addr = col.valuesNativeAddress()
+ while (i < count) {
+ sum += Platform.getInt(null, addr)
+ addr += 4
+ i += 1
+ }
+ }
+ col.close
+ }
+
+ // Access by going through a batch of unsafe rows.
+ val unsafeRowOnheap = { i: Int =>
+ val buffer = new Array[Byte](count * 16)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ val row = new UnsafeRow(1)
+ var i = 0
+ while (i < count) {
+ row.pointTo(buffer, Platform.BYTE_ARRAY_OFFSET + i * 16, 16)
+ row.setInt(0, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ row.pointTo(buffer, Platform.BYTE_ARRAY_OFFSET + i * 16, 16)
+ sum += row.getInt(0)
+ i += 1
+ }
+ }
+ }
+
+ // Access by going through a batch of unsafe rows.
+ val unsafeRowOffheap = { i: Int =>
+ val buffer = Platform.allocateMemory(count * 16)
+ var sum = 0L
+ for (n <- 0L until iters) {
+ val row = new UnsafeRow(1)
+ var i = 0
+ while (i < count) {
+ row.pointTo(null, buffer + i * 16, 16)
+ row.setInt(0, i)
+ i += 1
+ }
+ i = 0
+ while (i < count) {
+ row.pointTo(null, buffer + i * 16, 16)
+ sum += row.getInt(0)
+ i += 1
+ }
+ }
+ Platform.freeMemory(buffer)
+ }
+
+ /*
+ Intel(R) Core(TM) i7-4870HQ CPU @ 2.50GHz
+ Int Read/Write: Avg Time(ms) Avg Rate(M/s) Relative Rate
+ -------------------------------------------------------------------------
+ Java Array 248.8 1317.04 1.00 X
+ ByteBuffer Unsafe 435.6 752.25 0.57 X
+ ByteBuffer API 1752.0 187.03 0.14 X
+ DirectByteBuffer 595.4 550.35 0.42 X
+ Unsafe Buffer 235.2 1393.20 1.06 X
+ Column(on heap) 189.8 1726.45 1.31 X
--- End diff --
I'm really surprised that this one is faster than the Java Array, any clue?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49045676
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java ---
@@ -0,0 +1,165 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * A column backed by an in memory JVM array. This stores the NULLs as a byte per value
+ * and a java array for the values.
+ */
+public final class OnHeapColumnVector extends ColumnVector {
+ // The data stored in these arrays need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+
+ // This is faster than a boolean array and we optimize this over memory footprint.
+ private byte[] nulls;
+
+ // Array for each type. Only 1 is populated for any type.
+ private int[] intData;
+ private double[] doubleData;
+
+ protected OnHeapColumnVector(int capacity, DataType type) {
+ super(capacity, type);
+ if (type instanceof IntegerType) {
+ this.intData = new int[capacity];
+ } else if (type instanceof DoubleType) {
+ this.doubleData = new double[capacity];
+ } else {
+ throw new RuntimeException("Unhandled " + type);
+ }
+ this.nulls = new byte[capacity];
+ reset();
+ }
+
+ @Override
+ public final long valuesNativeAddress() {
+ throw new RuntimeException("Cannot get native address for on heap column");
+ }
+ @Override
+ public final long nullsNativeAddress() {
+ throw new RuntimeException("Cannot get native address for on heap column");
+ }
+
+ @Override
+ public final void close() {
+ nulls = null;
+ intData = null;
+ doubleData = null;
+ }
+
+
+ //
+ // APIs dealing with nulls
+ //
+
+ @Override
+ public final void putNotNull(int rowId) {
+ nulls[rowId] = (byte)0;
+ }
+
+ @Override
+ public final void putNull(int rowId) {
+ nulls[rowId] = (byte)1;
+ ++numNulls;
+ anyNullsSet = true;
+ }
+
+ @Override
+ public final void putNulls(int rowId, int count) {
+ for (int i = 0; i < count; ++i) {
+ nulls[rowId + i] = (byte)1;
+ }
+ anyNullsSet = true;
+ numNulls += count;
+ }
+
+ @Override
+ public final void putNotNulls(int rowId, int count) {
+ for (int i = 0; i < count; ++i) {
+ nulls[rowId + i] = (byte)0;
+ }
+ }
+
+ @Override
+ public final boolean getIsNull(int rowId) {
+ return nulls[rowId] == 1;
+ }
+
+ //
+ // APIs dealing with Ints
+ //
+
+ @Override
+ public final void putInt(int rowId, int value) {
+ intData[rowId] = value;
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int value) {
+ for (int i = 0; i < count; ++i) {
+ intData[i + rowId] = value;
+ }
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int[] src, int srcIndex) {
+ System.arraycopy(src, srcIndex, intData, rowId, count);
+ }
+
+ @Override
+ public final void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex) {
+ for (int i = 0; i < count; ++i) {
+ // TODO: is this the fastest way?
+ int v = (int)src[srcIndex];
+ v += ((int)src[srcIndex + 1]) << 8;
+ v += ((int)src[srcIndex + 2]) << 16;
+ v += ((int)src[srcIndex + 3]) << 24;
+ intData[i + rowId] = v;
+ srcIndex += 4;
+ }
+ }
+
+ @Override
+ public final int getInt(int rowId) {
+ return intData[rowId];
+ }
+
+ //
+ // APIs dealing with doubles
+ //
+
+ @Override
+ public final void putDouble(int rowId, double value) {
+ doubleData[rowId] = value;
+ }
+
+ @Override
+ public final void putDoubles(int rowId, int count, double value) {
+ for (int i = 0; i < count; ++i) {
+ doubleData[i + rowId] = value;
+ }
+ }
+
+ @Override
+ public final void putDoubles(int rowId, int count, double[] src, int srcIndex) {
+ System.arraycopy(src, srcIndex, doubleData, rowId, count);
+ }
+
+ @Override
+ public final void putDoubles(int rowId, int count, byte[] src, int srcIndex) {
+ DoubleBuffer values = ByteBuffer.wrap(src, srcIndex, src.length - srcIndex).asDoubleBuffer();
+ for (int i = 0; i < count; ++i) {
+ doubleData[i + rowId] = values.get(i);
--- End diff --
System.arrays.copy() ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49399720
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java ---
@@ -0,0 +1,165 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * A column backed by an in memory JVM array. This stores the NULLs as a byte per value
+ * and a java array for the values.
+ */
+public final class OnHeapColumnVector extends ColumnVector {
+ // The data stored in these arrays need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+
+ // This is faster than a boolean array and we optimize this over memory footprint.
+ private byte[] nulls;
+
+ // Array for each type. Only 1 is populated for any type.
+ private int[] intData;
+ private double[] doubleData;
+
+ protected OnHeapColumnVector(int capacity, DataType type) {
+ super(capacity, type);
+ if (type instanceof IntegerType) {
+ this.intData = new int[capacity];
+ } else if (type instanceof DoubleType) {
+ this.doubleData = new double[capacity];
+ } else {
+ throw new RuntimeException("Unhandled " + type);
+ }
+ this.nulls = new byte[capacity];
+ reset();
+ }
+
+ @Override
+ public final long valuesNativeAddress() {
+ throw new RuntimeException("Cannot get native address for on heap column");
+ }
+ @Override
+ public final long nullsNativeAddress() {
+ throw new RuntimeException("Cannot get native address for on heap column");
+ }
+
+ @Override
+ public final void close() {
+ nulls = null;
+ intData = null;
+ doubleData = null;
+ }
+
+
+ //
+ // APIs dealing with nulls
+ //
+
+ @Override
+ public final void putNotNull(int rowId) {
+ nulls[rowId] = (byte)0;
+ }
+
+ @Override
+ public final void putNull(int rowId) {
+ nulls[rowId] = (byte)1;
+ ++numNulls;
+ anyNullsSet = true;
+ }
+
+ @Override
+ public final void putNulls(int rowId, int count) {
+ for (int i = 0; i < count; ++i) {
+ nulls[rowId + i] = (byte)1;
+ }
+ anyNullsSet = true;
+ numNulls += count;
+ }
+
+ @Override
+ public final void putNotNulls(int rowId, int count) {
+ for (int i = 0; i < count; ++i) {
+ nulls[rowId + i] = (byte)0;
+ }
+ }
+
+ @Override
+ public final boolean getIsNull(int rowId) {
+ return nulls[rowId] == 1;
+ }
+
+ //
+ // APIs dealing with Ints
+ //
+
+ @Override
+ public final void putInt(int rowId, int value) {
+ intData[rowId] = value;
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int value) {
+ for (int i = 0; i < count; ++i) {
+ intData[i + rowId] = value;
+ }
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int[] src, int srcIndex) {
+ System.arraycopy(src, srcIndex, intData, rowId, count);
+ }
+
+ @Override
+ public final void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex) {
+ for (int i = 0; i < count; ++i) {
+ // TODO: is this the fastest way?
+ int v = (int)src[srcIndex];
+ v += ((int)src[srcIndex + 1]) << 8;
+ v += ((int)src[srcIndex + 2]) << 16;
+ v += ((int)src[srcIndex + 3]) << 24;
+ intData[i + rowId] = v;
+ srcIndex += 4;
+ }
+ }
+
+ @Override
+ public final int getInt(int rowId) {
+ return intData[rowId];
+ }
+
+ //
+ // APIs dealing with doubles
+ //
+
+ @Override
+ public final void putDouble(int rowId, double value) {
+ doubleData[rowId] = value;
+ }
+
+ @Override
+ public final void putDoubles(int rowId, int count, double value) {
+ for (int i = 0; i < count; ++i) {
+ doubleData[i + rowId] = value;
+ }
+ }
+
+ @Override
+ public final void putDoubles(int rowId, int count, double[] src, int srcIndex) {
+ System.arraycopy(src, srcIndex, doubleData, rowId, count);
+ }
+
+ @Override
+ public final void putDoubles(int rowId, int count, byte[] src, int srcIndex) {
+ DoubleBuffer values = ByteBuffer.wrap(src, srcIndex, src.length - srcIndex).asDoubleBuffer();
+ for (int i = 0; i < count; ++i) {
+ doubleData[i + rowId] = values.get(i);
--- End diff --
Does this work with mismatching types? Either way, we can do some more follow up to optimize these functions.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49392863
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java ---
@@ -0,0 +1,165 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * A column backed by an in memory JVM array. This stores the NULLs as a byte per value
+ * and a java array for the values.
+ */
+public final class OnHeapColumnVector extends ColumnVector {
+ // The data stored in these arrays need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+
+ // This is faster than a boolean array and we optimize this over memory footprint.
--- End diff --
Added a bitset/bytearray benchmark. Bytes are 50% faster. We can revisit though. For offheap in partiuclary, the popcnt instructions are useful.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
GitHub user nongli reopened a pull request:
https://github.com/apache/spark/pull/10628
[SPARK-12635][SQL] Add ColumnarBatch, an in memory columnar format for execution.
There are many potential benefits of having an efficient in memory columnar format as an alternate
to UnsafeRow. This patch introduces ColumnarBatch/ColumnarVector which starts this effort. The
remaining implementation can be done as follow up patches.
As stated in the in the JIRA, there are useful external components that operate on memory in a
simple columnar format. ColumnarBatch would serve that purpose and could server as a
zero-serialization/zero-copy exchange for this use case.
This patch supports running the underlying data either on heap or off heap. On heap runs a bit
faster but we would need offheap for zero-copy exchanges. Currently, this mode is hidden behind one
interface (ColumnVector).
This differs from Parquet or the existing columnar cache because this is *not* intended to be used
as a storage format. The focus is entirely on CPU efficiency as we expect to only have 1 of these
batches in memory per task. The layout of the values is just dense arrays of the value type.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/nongli/spark spark-12635
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/10628.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #10628
----
commit 84c657bdbc19105608e0d07fd59587b84eaaebcf
Author: Nong <no...@gmail.com>
Date: 2016-01-01T05:12:44Z
[SPARK-12635][SQL] Add ColumnarBatch, an in memory columnar format for execution.
There are many potential benefits of having an efficient in memory columnar format as an alternate
to UnsafeRow. This patch introduces ColumnarBatch/ColumnarVector which starts this effort. The
remaining implementation can be done as follow up patches.
As stated in the in the JIRA, there are useful external components that operate on memory in a
simple columnar format. ColumnarBatch would serve that purpose and could server as a
zero-serialization/zero-copy exchange for this use case.
This patch supports running the underlying data either on heap or off heap. On heap runs a bit
faster but we would need offheap for zero-copy exchanges. Currently, this mode is hidden behind one
interface (ColumnVector).
This differs from Parquet or the existing columnar cache because this is *not* intended to be used
as a storage format. The focus is entirely on CPU efficiency as we expect to only have 1 of these
batches in memory per task.
commit c547ec5df19fcd8dbd1becdd4fa98fd6246d5732
Author: Nong Li <no...@databricks.com>
Date: 2016-01-07T05:14:08Z
CR
commit 48574de403e05c3d3833356056eb88eb3c286d29
Author: Nong Li <no...@databricks.com>
Date: 2016-01-12T00:40:26Z
Fix double put api.
commit 91b6fc04fd6100ce14f1bab1fb2893464c79fbbf
Author: Nong Li <no...@databricks.com>
Date: 2016-01-12T02:06:30Z
Fix imports and rebase.
commit 36c8ddb6b51b3a7ee33950732287644a758615e7
Author: Nong Li <no...@databricks.com>
Date: 2016-01-12T05:52:08Z
Support java 7 iterator interface.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-170744337
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49042180
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java ---
@@ -0,0 +1,162 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * Column data backed using offheap memory.
+ */
+public final class OffHeapColumnVector extends ColumnVector {
+ // The data stored in these two allocations need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+ private long nulls;
+ private long data;
+
+ protected OffHeapColumnVector(int capacity, DataType type) {
+ super(capacity, type);
+ this.nulls = Platform.allocateMemory(capacity);
+ if (type instanceof IntegerType) {
+ this.data = Platform.allocateMemory(capacity * 4);
+ } else if (type instanceof DoubleType) {
+ this.data = Platform.allocateMemory(capacity * 8);
+ } else {
+ throw new RuntimeException("Unhandled " + type);
+ }
+ reset();
+ }
+
+ @Override
+ public final long valuesNativeAddress() {
+ return data;
+ }
+
+ @Override
+ public long nullsNativeAddress() {
+ return nulls;
+ }
+
+ @Override
+ public final void close() {
+ Platform.freeMemory(nulls);
+ Platform.freeMemory(data);
+ nulls = 0;
+ data = 0;
+ }
+
+ //
+ // APIs dealing with nulls
+ //
+
+ @Override
+ public final void putNotNull(int rowId) {
+ Platform.putByte(null, nulls + rowId, (byte) 0);
+ }
+
+ @Override
+ public final void putNull(int rowId) {
+ Platform.putByte(null, nulls + rowId, (byte) 1);
+ ++numNulls;
+ anyNullsSet = true;
+ }
+
+ @Override
+ public final void putNulls(int rowId, int count) {
+ long offset = nulls + rowId;
+ for (int i = 0; i < count; ++i, ++offset) {
+ Platform.putByte(null, offset, (byte) 1);
+ }
+ anyNullsSet = true;
+ numNulls += count;
+ }
+
+ @Override
+ public final void putNotNulls(int rowId, int count) {
+ long offset = nulls + rowId;
+ for (int i = 0; i < count; ++i, ++offset) {
+ Platform.putByte(null, offset, (byte) 0);
+ }
+ }
+
+ @Override
+ public final boolean getIsNull(int rowId) {
+ return Platform.getByte(null, nulls + rowId) == 1;
+ }
+
+ //
+ // APIs dealing with ints
+ //
+
+ @Override
+ public final void putInt(int rowId, int value) {
+ Platform.putInt(null, data + 4 * rowId, value);
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int value) {
+ long offset = data + 4 * rowId;
+ for (int i = 0; i < count; ++i, offset += 4) {
+ Platform.putInt(null, offset, value);
+ }
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int[] src, int srcIndex) {
+ Platform.copyMemory(src, Platform.INT_ARRAY_OFFSET + srcIndex * 4,
+ null, data + 4 * rowId, count * 4);
+ }
+
+ @Override
+ public final void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex) {
+ Platform.copyMemory(src, srcIndex + Platform.BYTE_ARRAY_OFFSET,
+ null, data + 4 * rowId, count * 4);
+ }
+
+ @Override
+ public final int getInt(int rowId) {
+ return Platform.getInt(null, data + 4 * rowId);
+ }
+
+ //
+ // APIs dealing with doubles
+ //
+
+ @Override
+ public final void putDouble(int rowId, double value) {
+ Platform.putDouble(null, data + rowId * 8, value);
+ }
+
+ @Override
+ public final void putDoubles(int rowId, int count, double value) {
+ long offset = data + 8 * rowId;
+ for (int i = 0; i < count; ++i, offset += 8) {
+ Platform.putDouble(null, offset, value);
+ }
+ }
+
+ @Override
+ public final void putDoubles(int rowId, int count, double[] src, int srcIndex) {
+ Platform.copyMemory(src, Platform.DOUBLE_ARRAY_OFFSET + srcIndex * 8,
+ null, data + 8 * rowId, count * 8);
+ }
+
+ @Override
+ public final void putDoubles(int rowId, int count, byte[] src, int srcIndex) {
+ // TODO: there must be a faster way to do this.
+ DoubleBuffer values = ByteBuffer.wrap(src, srcIndex, src.length - srcIndex).asDoubleBuffer();
--- End diff --
you can use unsafe to write this directly I think, assuming no change in endianness.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49044831
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarBatch.java ---
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.util.Iterator;
+
+import org.apache.commons.lang.NotImplementedException;
+
+/**
+ * This class is the in memory representation of rows as they are streamed through operators. It
+ * is designed to maximize CPU efficiency and not storage footprint. Since it is expected that
+ * each operator allocates one of thee objects, the storage footprint on the task is negligible.
+ *
+ * The layout is a columnar with values encoded in their native format. Each RowBatch contains
+ * a horizontal partitioning of the data, split into columns.
+ *
+ * The ColumnarBatch supports either on heap or offheap modes with (mostly) the identical API.
+ *
+ * TODO:
+ * - There are many TODOs for the existing APIs. They should throw a not implemented exception.
+ * - Compaction: The batch and columns should be able to compact based on a selection vector.
+ */
+public final class ColumnarBatch {
+ private static final int DEFAULT_BATCH_SIZE = 4 * 1024;
+
+ private final StructType schema;
+ private final int capacity;
+ private int numRows;
+ private final ColumnVector[] columns;
+
+ // True if the row is filtered.
+ private final boolean[] filteredRows;
+
+ // Total number of rows that have been filtered.
+ private int numRowsFiltered = 0;
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap) {
+ return new ColumnarBatch(schema, DEFAULT_BATCH_SIZE, offHeap);
+ }
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap, int maxRows) {
+ return new ColumnarBatch(schema, maxRows, offHeap);
+ }
+
+ /**
+ * Called to close all the columns in this batch. It is not valid to access the data after
+ * calling this. This must be called at the end to clean up memory allcoations.
+ */
+ public void close() {
+ for (ColumnVector c: columns) {
+ c.close();
+ }
+ }
+
+ /**
+ * Adapter class to interop with existing components that expect internal row. A lot of
+ * performance is lost with this translation.
+ */
+ public final class Row extends InternalRow {
+ private int rowId;
+
+ /**
+ * Marks this row as being filtered out. This means a subsequent iteration over the rows
+ * in this batch will not include this row.
+ */
+ public final void markFiltered() {
+ ColumnarBatch.this.markFiltered(rowId);
+ }
+
+ @Override
+ public final int numFields() {
+ return ColumnarBatch.this.numCols();
+ }
+
+ @Override
+ public final InternalRow copy() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final boolean anyNull() {
+ return false;
+ }
+
+ @Override
+ public final boolean isNullAt(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getIsNull(rowId);
+ }
+
+ @Override
+ public final boolean getBoolean(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final byte getByte(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final short getShort(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final int getInt(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getInt(rowId);
+ }
+
+ @Override
+ public final long getLong(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final float getFloat(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final double getDouble(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getDouble(rowId);
+ }
+
+ @Override
+ public final Decimal getDecimal(int ordinal, int precision, int scale) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final UTF8String getUTF8String(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final byte[] getBinary(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final CalendarInterval getInterval(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final InternalRow getStruct(int ordinal, int numFields) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final ArrayData getArray(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final MapData getMap(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final Object get(int ordinal, DataType dataType) {
+ throw new NotImplementedException();
+ }
+ }
+
+ /**
+ * Returns an iterator over the rows in this batch. This skips rows that are filtered out.
+ */
+ public Iterator<Row> rowIterator() {
+ final int maxRows = ColumnarBatch.this.numRows();
+ final Row row = new Row();
+ return new Iterator<Row>() {
+ int rowId = 0;
+
+ @Override
+ public boolean hasNext() {
+ while (rowId < maxRows && ColumnarBatch.this.filteredRows[rowId]) {
+ ++rowId;
+ }
+ return rowId < maxRows;
+ }
+
+ @Override
+ public Row next() {
+ while (rowId < maxRows && ColumnarBatch.this.filteredRows[rowId]) {
+ ++rowId;
+ }
+ row.rowId = rowId++;
+ return row;
+ }
+ };
+ }
+
+ /**
+ * Resets the batch for writing.
+ */
+ public void reset() {
+ for (int i = 0; i < numCols(); ++i) {
+ columns[i].reset();
+ }
+ if (this.numRowsFiltered > 0) {
+ for (int i = 0; i < numRows; ++i) {
+ filteredRows[i] = false;
--- End diff --
Will System.arrays.fill() be faster than this loop?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12785][SQL] Add ColumnarBatch, an in me...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-171135786
I'm going to merge this. Thanks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49044960
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarBatch.java ---
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.util.Iterator;
+
+import org.apache.commons.lang.NotImplementedException;
+
+/**
+ * This class is the in memory representation of rows as they are streamed through operators. It
+ * is designed to maximize CPU efficiency and not storage footprint. Since it is expected that
+ * each operator allocates one of thee objects, the storage footprint on the task is negligible.
+ *
+ * The layout is a columnar with values encoded in their native format. Each RowBatch contains
+ * a horizontal partitioning of the data, split into columns.
+ *
+ * The ColumnarBatch supports either on heap or offheap modes with (mostly) the identical API.
+ *
+ * TODO:
+ * - There are many TODOs for the existing APIs. They should throw a not implemented exception.
+ * - Compaction: The batch and columns should be able to compact based on a selection vector.
+ */
+public final class ColumnarBatch {
+ private static final int DEFAULT_BATCH_SIZE = 4 * 1024;
+
+ private final StructType schema;
+ private final int capacity;
--- End diff --
Should we support growing when needed? A small batch could produce very big batch (for example, in join), it's hard to have a proper hard limit.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli closed the pull request at:
https://github.com/apache/spark/pull/10628
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-170828321
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49376528
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarBatch.java ---
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.util.Iterator;
+
+import org.apache.commons.lang.NotImplementedException;
+
+/**
+ * This class is the in memory representation of rows as they are streamed through operators. It
+ * is designed to maximize CPU efficiency and not storage footprint. Since it is expected that
+ * each operator allocates one of thee objects, the storage footprint on the task is negligible.
+ *
+ * The layout is a columnar with values encoded in their native format. Each RowBatch contains
+ * a horizontal partitioning of the data, split into columns.
+ *
+ * The ColumnarBatch supports either on heap or offheap modes with (mostly) the identical API.
+ *
+ * TODO:
+ * - There are many TODOs for the existing APIs. They should throw a not implemented exception.
+ * - Compaction: The batch and columns should be able to compact based on a selection vector.
+ */
+public final class ColumnarBatch {
+ private static final int DEFAULT_BATCH_SIZE = 4 * 1024;
+
+ private final StructType schema;
+ private final int capacity;
+ private int numRows;
+ private final ColumnVector[] columns;
+
+ // True if the row is filtered.
+ private final boolean[] filteredRows;
+
+ // Total number of rows that have been filtered.
+ private int numRowsFiltered = 0;
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap) {
+ return new ColumnarBatch(schema, DEFAULT_BATCH_SIZE, offHeap);
+ }
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap, int maxRows) {
+ return new ColumnarBatch(schema, maxRows, offHeap);
+ }
+
+ /**
+ * Called to close all the columns in this batch. It is not valid to access the data after
+ * calling this. This must be called at the end to clean up memory allcoations.
+ */
+ public void close() {
+ for (ColumnVector c: columns) {
+ c.close();
+ }
+ }
+
+ /**
+ * Adapter class to interop with existing components that expect internal row. A lot of
+ * performance is lost with this translation.
+ */
+ public final class Row extends InternalRow {
+ private int rowId;
+
+ /**
+ * Marks this row as being filtered out. This means a subsequent iteration over the rows
+ * in this batch will not include this row.
+ */
+ public final void markFiltered() {
+ ColumnarBatch.this.markFiltered(rowId);
+ }
+
+ @Override
+ public final int numFields() {
+ return ColumnarBatch.this.numCols();
+ }
+
+ @Override
+ public final InternalRow copy() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final boolean anyNull() {
+ return false;
+ }
+
+ @Override
+ public final boolean isNullAt(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getIsNull(rowId);
+ }
+
+ @Override
+ public final boolean getBoolean(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final byte getByte(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final short getShort(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final int getInt(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getInt(rowId);
+ }
+
+ @Override
+ public final long getLong(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final float getFloat(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final double getDouble(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getDouble(rowId);
+ }
+
+ @Override
+ public final Decimal getDecimal(int ordinal, int precision, int scale) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final UTF8String getUTF8String(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final byte[] getBinary(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final CalendarInterval getInterval(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final InternalRow getStruct(int ordinal, int numFields) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final ArrayData getArray(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final MapData getMap(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final Object get(int ordinal, DataType dataType) {
+ throw new NotImplementedException();
+ }
+ }
+
+ /**
+ * Returns an iterator over the rows in this batch. This skips rows that are filtered out.
+ */
+ public Iterator<Row> rowIterator() {
+ final int maxRows = ColumnarBatch.this.numRows();
+ final Row row = new Row();
+ return new Iterator<Row>() {
+ int rowId = 0;
+
+ @Override
+ public boolean hasNext() {
+ while (rowId < maxRows && ColumnarBatch.this.filteredRows[rowId]) {
+ ++rowId;
+ }
+ return rowId < maxRows;
+ }
+
+ @Override
+ public Row next() {
+ while (rowId < maxRows && ColumnarBatch.this.filteredRows[rowId]) {
+ ++rowId;
+ }
+ row.rowId = rowId++;
+ return row;
+ }
+ };
+ }
+
+ /**
+ * Resets the batch for writing.
+ */
+ public void reset() {
+ for (int i = 0; i < numCols(); ++i) {
+ columns[i].reset();
+ }
+ if (this.numRowsFiltered > 0) {
+ for (int i = 0; i < numRows; ++i) {
+ filteredRows[i] = false;
--- End diff --
Didn't benchmark it but it seems clearer. This thing is not off heap.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-170808713
**[Test build #49216 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/49216/consoleFull)** for PR 10628 at commit [`36c8ddb`](https://github.com/apache/spark/commit/36c8ddb6b51b3a7ee33950732287644a758615e7).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49044948
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarBatch.java ---
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.util.Iterator;
+
+import org.apache.commons.lang.NotImplementedException;
+
+/**
+ * This class is the in memory representation of rows as they are streamed through operators. It
+ * is designed to maximize CPU efficiency and not storage footprint. Since it is expected that
+ * each operator allocates one of thee objects, the storage footprint on the task is negligible.
+ *
+ * The layout is a columnar with values encoded in their native format. Each RowBatch contains
+ * a horizontal partitioning of the data, split into columns.
+ *
+ * The ColumnarBatch supports either on heap or offheap modes with (mostly) the identical API.
+ *
+ * TODO:
+ * - There are many TODOs for the existing APIs. They should throw a not implemented exception.
+ * - Compaction: The batch and columns should be able to compact based on a selection vector.
+ */
+public final class ColumnarBatch {
+ private static final int DEFAULT_BATCH_SIZE = 4 * 1024;
+
+ private final StructType schema;
+ private final int capacity;
+ private int numRows;
+ private final ColumnVector[] columns;
+
+ // True if the row is filtered.
+ private final boolean[] filteredRows;
+
+ // Total number of rows that have been filtered.
+ private int numRowsFiltered = 0;
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap) {
+ return new ColumnarBatch(schema, DEFAULT_BATCH_SIZE, offHeap);
+ }
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap, int maxRows) {
+ return new ColumnarBatch(schema, maxRows, offHeap);
+ }
+
+ /**
+ * Called to close all the columns in this batch. It is not valid to access the data after
+ * calling this. This must be called at the end to clean up memory allcoations.
+ */
+ public void close() {
+ for (ColumnVector c: columns) {
+ c.close();
+ }
+ }
+
+ /**
+ * Adapter class to interop with existing components that expect internal row. A lot of
+ * performance is lost with this translation.
+ */
+ public final class Row extends InternalRow {
+ private int rowId;
+
+ /**
+ * Marks this row as being filtered out. This means a subsequent iteration over the rows
+ * in this batch will not include this row.
+ */
+ public final void markFiltered() {
+ ColumnarBatch.this.markFiltered(rowId);
+ }
+
+ @Override
+ public final int numFields() {
+ return ColumnarBatch.this.numCols();
+ }
+
+ @Override
+ public final InternalRow copy() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final boolean anyNull() {
+ return false;
+ }
+
+ @Override
+ public final boolean isNullAt(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getIsNull(rowId);
+ }
+
+ @Override
+ public final boolean getBoolean(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final byte getByte(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final short getShort(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final int getInt(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getInt(rowId);
+ }
+
+ @Override
+ public final long getLong(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final float getFloat(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final double getDouble(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getDouble(rowId);
+ }
+
+ @Override
+ public final Decimal getDecimal(int ordinal, int precision, int scale) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final UTF8String getUTF8String(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final byte[] getBinary(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final CalendarInterval getInterval(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final InternalRow getStruct(int ordinal, int numFields) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final ArrayData getArray(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final MapData getMap(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final Object get(int ordinal, DataType dataType) {
+ throw new NotImplementedException();
+ }
+ }
+
+ /**
+ * Returns an iterator over the rows in this batch. This skips rows that are filtered out.
+ */
+ public Iterator<Row> rowIterator() {
+ final int maxRows = ColumnarBatch.this.numRows();
+ final Row row = new Row();
+ return new Iterator<Row>() {
+ int rowId = 0;
+
+ @Override
+ public boolean hasNext() {
+ while (rowId < maxRows && ColumnarBatch.this.filteredRows[rowId]) {
+ ++rowId;
+ }
+ return rowId < maxRows;
+ }
+
+ @Override
+ public Row next() {
+ while (rowId < maxRows && ColumnarBatch.this.filteredRows[rowId]) {
+ ++rowId;
+ }
+ row.rowId = rowId++;
+ return row;
+ }
+ };
+ }
+
+ /**
+ * Resets the batch for writing.
+ */
+ public void reset() {
+ for (int i = 0; i < numCols(); ++i) {
+ columns[i].reset();
+ }
+ if (this.numRowsFiltered > 0) {
+ for (int i = 0; i < numRows; ++i) {
+ filteredRows[i] = false;
--- End diff --
Use this http://www.docjar.com/docs/api/sun/misc/Unsafe.html#setMemory(long, long, byte)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49044780
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarBatch.java ---
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.util.Iterator;
+
+import org.apache.commons.lang.NotImplementedException;
+
+/**
+ * This class is the in memory representation of rows as they are streamed through operators. It
+ * is designed to maximize CPU efficiency and not storage footprint. Since it is expected that
+ * each operator allocates one of thee objects, the storage footprint on the task is negligible.
+ *
+ * The layout is a columnar with values encoded in their native format. Each RowBatch contains
+ * a horizontal partitioning of the data, split into columns.
+ *
+ * The ColumnarBatch supports either on heap or offheap modes with (mostly) the identical API.
+ *
+ * TODO:
+ * - There are many TODOs for the existing APIs. They should throw a not implemented exception.
+ * - Compaction: The batch and columns should be able to compact based on a selection vector.
+ */
+public final class ColumnarBatch {
+ private static final int DEFAULT_BATCH_SIZE = 4 * 1024;
+
+ private final StructType schema;
+ private final int capacity;
+ private int numRows;
+ private final ColumnVector[] columns;
+
+ // True if the row is filtered.
+ private final boolean[] filteredRows;
+
+ // Total number of rows that have been filtered.
+ private int numRowsFiltered = 0;
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap) {
+ return new ColumnarBatch(schema, DEFAULT_BATCH_SIZE, offHeap);
+ }
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap, int maxRows) {
+ return new ColumnarBatch(schema, maxRows, offHeap);
+ }
+
+ /**
+ * Called to close all the columns in this batch. It is not valid to access the data after
+ * calling this. This must be called at the end to clean up memory allcoations.
+ */
+ public void close() {
+ for (ColumnVector c: columns) {
+ c.close();
+ }
+ }
+
+ /**
+ * Adapter class to interop with existing components that expect internal row. A lot of
+ * performance is lost with this translation.
+ */
+ public final class Row extends InternalRow {
+ private int rowId;
+
+ /**
+ * Marks this row as being filtered out. This means a subsequent iteration over the rows
+ * in this batch will not include this row.
+ */
+ public final void markFiltered() {
+ ColumnarBatch.this.markFiltered(rowId);
+ }
+
+ @Override
+ public final int numFields() {
+ return ColumnarBatch.this.numCols();
+ }
+
+ @Override
+ public final InternalRow copy() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final boolean anyNull() {
+ return false;
+ }
+
+ @Override
+ public final boolean isNullAt(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getIsNull(rowId);
+ }
+
+ @Override
+ public final boolean getBoolean(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final byte getByte(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final short getShort(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final int getInt(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getInt(rowId);
+ }
+
+ @Override
+ public final long getLong(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final float getFloat(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final double getDouble(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getDouble(rowId);
+ }
+
+ @Override
+ public final Decimal getDecimal(int ordinal, int precision, int scale) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final UTF8String getUTF8String(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final byte[] getBinary(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final CalendarInterval getInterval(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final InternalRow getStruct(int ordinal, int numFields) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final ArrayData getArray(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final MapData getMap(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final Object get(int ordinal, DataType dataType) {
+ throw new NotImplementedException();
+ }
+ }
+
+ /**
+ * Returns an iterator over the rows in this batch. This skips rows that are filtered out.
+ */
+ public Iterator<Row> rowIterator() {
+ final int maxRows = ColumnarBatch.this.numRows();
+ final Row row = new Row();
+ return new Iterator<Row>() {
+ int rowId = 0;
+
+ @Override
+ public boolean hasNext() {
+ while (rowId < maxRows && ColumnarBatch.this.filteredRows[rowId]) {
+ ++rowId;
+ }
+ return rowId < maxRows;
+ }
+
+ @Override
+ public Row next() {
+ while (rowId < maxRows && ColumnarBatch.this.filteredRows[rowId]) {
+ ++rowId;
+ }
+ row.rowId = rowId++;
--- End diff --
Should we throw a exception if rowId == maxRows?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49382888
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java ---
@@ -0,0 +1,162 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * Column data backed using offheap memory.
+ */
+public final class OffHeapColumnVector extends ColumnVector {
+ // The data stored in these two allocations need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+ private long nulls;
+ private long data;
+
+ protected OffHeapColumnVector(int capacity, DataType type) {
+ super(capacity, type);
+ this.nulls = Platform.allocateMemory(capacity);
+ if (type instanceof IntegerType) {
+ this.data = Platform.allocateMemory(capacity * 4);
+ } else if (type instanceof DoubleType) {
+ this.data = Platform.allocateMemory(capacity * 8);
+ } else {
+ throw new RuntimeException("Unhandled " + type);
+ }
+ reset();
+ }
+
+ @Override
+ public final long valuesNativeAddress() {
+ return data;
+ }
+
+ @Override
+ public long nullsNativeAddress() {
+ return nulls;
+ }
+
+ @Override
+ public final void close() {
+ Platform.freeMemory(nulls);
+ Platform.freeMemory(data);
+ nulls = 0;
+ data = 0;
+ }
+
+ //
+ // APIs dealing with nulls
+ //
+
+ @Override
+ public final void putNotNull(int rowId) {
+ Platform.putByte(null, nulls + rowId, (byte) 0);
+ }
+
+ @Override
+ public final void putNull(int rowId) {
+ Platform.putByte(null, nulls + rowId, (byte) 1);
+ ++numNulls;
+ anyNullsSet = true;
+ }
+
+ @Override
+ public final void putNulls(int rowId, int count) {
+ long offset = nulls + rowId;
+ for (int i = 0; i < count; ++i, ++offset) {
+ Platform.putByte(null, offset, (byte) 1);
+ }
+ anyNullsSet = true;
+ numNulls += count;
+ }
+
+ @Override
+ public final void putNotNulls(int rowId, int count) {
+ long offset = nulls + rowId;
+ for (int i = 0; i < count; ++i, ++offset) {
+ Platform.putByte(null, offset, (byte) 0);
+ }
+ }
+
+ @Override
+ public final boolean getIsNull(int rowId) {
+ return Platform.getByte(null, nulls + rowId) == 1;
+ }
+
+ //
+ // APIs dealing with ints
+ //
+
+ @Override
+ public final void putInt(int rowId, int value) {
+ Platform.putInt(null, data + 4 * rowId, value);
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int value) {
+ long offset = data + 4 * rowId;
+ for (int i = 0; i < count; ++i, offset += 4) {
+ Platform.putInt(null, offset, value);
+ }
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int[] src, int srcIndex) {
+ Platform.copyMemory(src, Platform.INT_ARRAY_OFFSET + srcIndex * 4,
+ null, data + 4 * rowId, count * 4);
+ }
+
+ @Override
+ public final void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex) {
--- End diff --
Then we should check the endianness here, or this will broken on BigEndian machines
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-170746066
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49400006
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarBatch.java ---
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.util.Iterator;
+
+import org.apache.commons.lang.NotImplementedException;
+
+/**
+ * This class is the in memory representation of rows as they are streamed through operators. It
+ * is designed to maximize CPU efficiency and not storage footprint. Since it is expected that
+ * each operator allocates one of thee objects, the storage footprint on the task is negligible.
+ *
+ * The layout is a columnar with values encoded in their native format. Each RowBatch contains
+ * a horizontal partitioning of the data, split into columns.
+ *
+ * The ColumnarBatch supports either on heap or offheap modes with (mostly) the identical API.
+ *
+ * TODO:
+ * - There are many TODOs for the existing APIs. They should throw a not implemented exception.
+ * - Compaction: The batch and columns should be able to compact based on a selection vector.
+ */
+public final class ColumnarBatch {
+ private static final int DEFAULT_BATCH_SIZE = 4 * 1024;
+
+ private final StructType schema;
+ private final int capacity;
+ private int numRows;
+ private final ColumnVector[] columns;
+
+ // True if the row is filtered.
+ private final boolean[] filteredRows;
+
+ // Total number of rows that have been filtered.
+ private int numRowsFiltered = 0;
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap) {
+ return new ColumnarBatch(schema, DEFAULT_BATCH_SIZE, offHeap);
+ }
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap, int maxRows) {
+ return new ColumnarBatch(schema, maxRows, offHeap);
+ }
+
+ /**
+ * Called to close all the columns in this batch. It is not valid to access the data after
+ * calling this. This must be called at the end to clean up memory allcoations.
+ */
+ public void close() {
+ for (ColumnVector c: columns) {
+ c.close();
+ }
+ }
+
+ /**
+ * Adapter class to interop with existing components that expect internal row. A lot of
+ * performance is lost with this translation.
+ */
+ public final class Row extends InternalRow {
+ private int rowId;
+
+ /**
+ * Marks this row as being filtered out. This means a subsequent iteration over the rows
+ * in this batch will not include this row.
+ */
+ public final void markFiltered() {
+ ColumnarBatch.this.markFiltered(rowId);
+ }
+
+ @Override
+ public final int numFields() {
+ return ColumnarBatch.this.numCols();
+ }
+
+ @Override
+ public final InternalRow copy() {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final boolean anyNull() {
+ return false;
+ }
+
+ @Override
+ public final boolean isNullAt(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getIsNull(rowId);
+ }
+
+ @Override
+ public final boolean getBoolean(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final byte getByte(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final short getShort(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final int getInt(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getInt(rowId);
+ }
+
+ @Override
+ public final long getLong(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final float getFloat(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final double getDouble(int ordinal) {
+ return ColumnarBatch.this.column(ordinal).getDouble(rowId);
+ }
+
+ @Override
+ public final Decimal getDecimal(int ordinal, int precision, int scale) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final UTF8String getUTF8String(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final byte[] getBinary(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final CalendarInterval getInterval(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final InternalRow getStruct(int ordinal, int numFields) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final ArrayData getArray(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final MapData getMap(int ordinal) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public final Object get(int ordinal, DataType dataType) {
+ throw new NotImplementedException();
+ }
+ }
+
+ /**
+ * Returns an iterator over the rows in this batch. This skips rows that are filtered out.
+ */
+ public Iterator<Row> rowIterator() {
+ final int maxRows = ColumnarBatch.this.numRows();
+ final Row row = new Row();
+ return new Iterator<Row>() {
+ int rowId = 0;
+
+ @Override
+ public boolean hasNext() {
+ while (rowId < maxRows && ColumnarBatch.this.filteredRows[rowId]) {
+ ++rowId;
+ }
+ return rowId < maxRows;
+ }
+
+ @Override
+ public Row next() {
+ while (rowId < maxRows && ColumnarBatch.this.filteredRows[rowId]) {
+ ++rowId;
+ }
+ row.rowId = rowId++;
+ return row;
+ }
+ };
+ }
+
+ /**
+ * Resets the batch for writing.
+ */
+ public void reset() {
+ for (int i = 0; i < numCols(); ++i) {
+ columns[i].reset();
+ }
+ if (this.numRowsFiltered > 0) {
+ for (int i = 0; i < numRows; ++i) {
+ filteredRows[i] = false;
--- End diff --
Yea - there is also a variant of it that can take an object with an offset.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49045516
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java ---
@@ -0,0 +1,165 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * A column backed by an in memory JVM array. This stores the NULLs as a byte per value
+ * and a java array for the values.
+ */
+public final class OnHeapColumnVector extends ColumnVector {
+ // The data stored in these arrays need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+
+ // This is faster than a boolean array and we optimize this over memory footprint.
--- End diff --
UnsafeRow should also use one byte per column (only 10% if the number of column is larger than 8), cc @rxin
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-169551685
This patch uses this class to decode parquet.
https://github.com/apache/spark/pull/10593
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12785][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-205993243
I don't think it should be needed for OnHeap (it uses Platform.getInt which I think should work for both). For offheap, as far as I know, only a few of the puts that take byte[] as input would care. Subclassing or if(...) both sound fine to me.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-169552400
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/48897/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-170746058
**[Test build #49195 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/49195/consoleFull)** for PR 10628 at commit [`5b27f65`](https://github.com/apache/spark/commit/5b27f651db7b4bab17d0a33d472bf47178b52390).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49045016
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java ---
@@ -0,0 +1,162 @@
+package org.apache.spark.sql.execution.vectorized;
--- End diff --
License header
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-169552398
**[Test build #48897 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/48897/consoleFull)** for PR 10628 at commit [`381492e`](https://github.com/apache/spark/commit/381492eee072a8e1a536103db1e6afcff008b6b1).
* This patch **fails RAT tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `public abstract class ColumnVector `
* `public final class ColumnarBatch `
* ` public final class Row extends InternalRow `
* `public final class OffHeapColumnVector extends ColumnVector `
* `public final class OnHeapColumnVector extends ColumnVector `
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-170746070
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/49195/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-170827526
**[Test build #49216 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/49216/consoleFull)** for PR 10628 at commit [`36c8ddb`](https://github.com/apache/spark/commit/36c8ddb6b51b3a7ee33950732287644a758615e7).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49044595
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnarBatch.java ---
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.unsafe.types.CalendarInterval;
+import org.apache.spark.unsafe.types.UTF8String;
+
+import java.util.Iterator;
+
+import org.apache.commons.lang.NotImplementedException;
+
+/**
+ * This class is the in memory representation of rows as they are streamed through operators. It
+ * is designed to maximize CPU efficiency and not storage footprint. Since it is expected that
+ * each operator allocates one of thee objects, the storage footprint on the task is negligible.
+ *
+ * The layout is a columnar with values encoded in their native format. Each RowBatch contains
+ * a horizontal partitioning of the data, split into columns.
+ *
+ * The ColumnarBatch supports either on heap or offheap modes with (mostly) the identical API.
+ *
+ * TODO:
+ * - There are many TODOs for the existing APIs. They should throw a not implemented exception.
+ * - Compaction: The batch and columns should be able to compact based on a selection vector.
+ */
+public final class ColumnarBatch {
+ private static final int DEFAULT_BATCH_SIZE = 4 * 1024;
+
+ private final StructType schema;
+ private final int capacity;
+ private int numRows;
+ private final ColumnVector[] columns;
+
+ // True if the row is filtered.
+ private final boolean[] filteredRows;
+
+ // Total number of rows that have been filtered.
+ private int numRowsFiltered = 0;
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap) {
+ return new ColumnarBatch(schema, DEFAULT_BATCH_SIZE, offHeap);
+ }
+
+ public static ColumnarBatch allocate(StructType schema, boolean offHeap, int maxRows) {
+ return new ColumnarBatch(schema, maxRows, offHeap);
+ }
+
+ /**
+ * Called to close all the columns in this batch. It is not valid to access the data after
+ * calling this. This must be called at the end to clean up memory allcoations.
+ */
+ public void close() {
+ for (ColumnVector c: columns) {
+ c.close();
+ }
+ }
+
+ /**
+ * Adapter class to interop with existing components that expect internal row. A lot of
+ * performance is lost with this translation.
+ */
+ public final class Row extends InternalRow {
+ private int rowId;
+
+ /**
+ * Marks this row as being filtered out. This means a subsequent iteration over the rows
+ * in this batch will not include this row.
+ */
+ public final void markFiltered() {
+ ColumnarBatch.this.markFiltered(rowId);
+ }
+
+ @Override
+ public final int numFields() {
+ return ColumnarBatch.this.numCols();
+ }
+
+ @Override
+ public final InternalRow copy() {
+ throw new NotImplementedException();
--- End diff --
We could generate a UnsafeRow for this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-169552335
**[Test build #48897 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/48897/consoleFull)** for PR 10628 at commit [`381492e`](https://github.com/apache/spark/commit/381492eee072a8e1a536103db1e6afcff008b6b1).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49377063
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java ---
@@ -0,0 +1,162 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * Column data backed using offheap memory.
+ */
+public final class OffHeapColumnVector extends ColumnVector {
+ // The data stored in these two allocations need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+ private long nulls;
+ private long data;
+
+ protected OffHeapColumnVector(int capacity, DataType type) {
+ super(capacity, type);
+ this.nulls = Platform.allocateMemory(capacity);
--- End diff --
Whats the benefit of having them back to back? These arrays are way too big to to fit in cache lines.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-170762755
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/49201/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-170828330
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/49216/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12785][SQL] Add ColumnarBatch, an in me...
Posted by a-roberts <gi...@git.apache.org>.
Github user a-roberts commented on the pull request:
https://github.com/apache/spark/pull/10628#issuecomment-204397652
@nongli @davies
Hi, by implementing the feature this way we've eliminated every big-endian user from using this feature with Spark - Z systems, for example, or Power big-endian machines. Critical if we want Spark to be used by large industries such as those that IBM et al support.
Any intention to have this not rely on the byte ordering of the platform? putLittleEndian is not going to cut it when we're wanting to use Spark for handling large volumes of customer data that traditionally sits on big-endian systems.
Also, are there any design docs for this? I found the documentation for Tungsten (UnsafeRows) to be very useful for understanding and debugging purposes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by nongli <gi...@git.apache.org>.
Github user nongli commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49399204
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java ---
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+
+/**
+ * This class represents a column of values and provides the main APIs to access the data
+ * values. It supports all the types and contains get/put APIs as well as their batched versions.
+ * The batched versions are preferable whenever possible.
+ *
+ * Most of the APIs take the rowId as a parameter. This is the local 0-based row id for values
+ * in the current RowBatch.
+ *
+ * A ColumnVector should be considered immutable once originally created. In other words, it is not
+ * valid to call put APIs after reads until reset() is called.
+ */
+public abstract class ColumnVector {
+ /**
+ * Allocates a column with each element of size `width` either on or off heap.
+ */
+ public static ColumnVector allocate(int capacity, DataType type, boolean offHeap) {
+ if (offHeap) {
+ return new OffHeapColumnVector(capacity, type);
+ } else {
+ return new OnHeapColumnVector(capacity, type);
+ }
+ }
+
+ public final DataType dataType() { return type; }
+
+ /**
+ * Resets this column for writing. The currently stored values are no longer accessible.
+ */
+ public void reset() {
+ numNulls = 0;
+ if (anyNullsSet) {
+ putNotNulls(0, capacity);
+ anyNullsSet = false;
+ }
+ }
+
+ /**
+ * Cleans up memory for this column. The column is not usable after this.
+ * TODO: this should probably have ref-counted semantics.
+ */
+ public abstract void close();
+
+ /**
+ * Returns the number of nulls in this column.
+ */
+ public final int numNulls() { return numNulls; }
+
+ /**
+ * Returns true if any of the nulls indicator are set for this column. This can be used
+ * as an optimization to prevent setting nulls.
+ */
+ public final boolean anyNullsSet() { return anyNullsSet; }
+
+ /**
+ * Returns the off heap ptr for the arrays backing the NULLs and values buffer. Only valid
+ * to call for off heap columns.
+ */
+ public abstract long nullsNativeAddress();
+ public abstract long valuesNativeAddress();
+
+ /**
+ * Sets the value at rowId to null/not null.
+ */
+ public abstract void putNotNull(int rowId);
+ public abstract void putNull(int rowId);
+
+ /**
+ * Sets the values from [rowId, rowId + count) to null/not null.
+ */
+ public abstract void putNulls(int rowId, int count);
+ public abstract void putNotNulls(int rowId, int count);
+
+ /**
+ * Returns whether the value at rowId is NULL.
+ */
+ public abstract boolean getIsNull(int rowId);
+
+ /**
+ * Sets the value at rowId to `value`.
+ */
+ public abstract void putInt(int rowId, int value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to value.
+ */
+ public abstract void putInts(int rowId, int count, int value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ */
+ public abstract void putInts(int rowId, int count, int[] src, int srcIndex);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * The data in src must be 4-byte little endian ints.
+ */
+ public abstract void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex);
+
+ /**
+ * Returns the integer for rowId.
+ */
+ public abstract int getInt(int rowId);
+
+ /**
+ * Sets the value at rowId to `value`.
+ */
+ public abstract void putDouble(int rowId, double value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to value.
+ */
+ public abstract void putDoubles(int rowId, int count, double value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * src should contain `count` doubles written as ieee format.
+ */
+ public abstract void putDoubles(int rowId, int count, double[] src, int srcIndex);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * The data in src must be ieee formated doubles.
+ */
+ public abstract void putDoubles(int rowId, int count, byte[] src, int srcIndex);
+
+ /**
+ * Returns the double for rowId.
+ */
+ public abstract double getDouble(int rowId);
+
+ /**
+ * Maximum number of rows that can be stored in this column.
+ */
+ protected final int capacity;
+
+ /**
+ * Byte width fo this column.
+ */
+ protected final int width;
+
+ /**
+ * Number of nulls in this column.
+ */
+ protected int numNulls;
+
+ /**
+ * True if there is at least one NULL byte set.
+ */
+ protected boolean anyNullsSet;
+
+ /**
+ * Data type for this column.
+ */
+ protected final DataType type;
+
+ protected ColumnVector(int capacity, DataType type) {
+ this.capacity = capacity;
+ this.type = type;
+ this.width = type.defaultSize();
+ }
+}
--- End diff --
I'd put that as processing on encoded data. We can add that in the future for sure.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49045224
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java ---
@@ -0,0 +1,162 @@
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.unsafe.Platform;
+
+import java.nio.ByteBuffer;
+import java.nio.DoubleBuffer;
+
+/**
+ * Column data backed using offheap memory.
+ */
+public final class OffHeapColumnVector extends ColumnVector {
+ // The data stored in these two allocations need to maintain binary compatible. We can
+ // directly pass this buffer to external components.
+ private long nulls;
+ private long data;
+
+ protected OffHeapColumnVector(int capacity, DataType type) {
+ super(capacity, type);
+ this.nulls = Platform.allocateMemory(capacity);
+ if (type instanceof IntegerType) {
+ this.data = Platform.allocateMemory(capacity * 4);
+ } else if (type instanceof DoubleType) {
+ this.data = Platform.allocateMemory(capacity * 8);
+ } else {
+ throw new RuntimeException("Unhandled " + type);
+ }
+ reset();
+ }
+
+ @Override
+ public final long valuesNativeAddress() {
+ return data;
+ }
+
+ @Override
+ public long nullsNativeAddress() {
+ return nulls;
+ }
+
+ @Override
+ public final void close() {
+ Platform.freeMemory(nulls);
+ Platform.freeMemory(data);
+ nulls = 0;
+ data = 0;
+ }
+
+ //
+ // APIs dealing with nulls
+ //
+
+ @Override
+ public final void putNotNull(int rowId) {
+ Platform.putByte(null, nulls + rowId, (byte) 0);
+ }
+
+ @Override
+ public final void putNull(int rowId) {
+ Platform.putByte(null, nulls + rowId, (byte) 1);
+ ++numNulls;
+ anyNullsSet = true;
+ }
+
+ @Override
+ public final void putNulls(int rowId, int count) {
+ long offset = nulls + rowId;
+ for (int i = 0; i < count; ++i, ++offset) {
+ Platform.putByte(null, offset, (byte) 1);
+ }
+ anyNullsSet = true;
+ numNulls += count;
+ }
+
+ @Override
+ public final void putNotNulls(int rowId, int count) {
+ long offset = nulls + rowId;
+ for (int i = 0; i < count; ++i, ++offset) {
+ Platform.putByte(null, offset, (byte) 0);
+ }
+ }
+
+ @Override
+ public final boolean getIsNull(int rowId) {
+ return Platform.getByte(null, nulls + rowId) == 1;
+ }
+
+ //
+ // APIs dealing with ints
+ //
+
+ @Override
+ public final void putInt(int rowId, int value) {
+ Platform.putInt(null, data + 4 * rowId, value);
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int value) {
+ long offset = data + 4 * rowId;
+ for (int i = 0; i < count; ++i, offset += 4) {
+ Platform.putInt(null, offset, value);
+ }
+ }
+
+ @Override
+ public final void putInts(int rowId, int count, int[] src, int srcIndex) {
+ Platform.copyMemory(src, Platform.INT_ARRAY_OFFSET + srcIndex * 4,
+ null, data + 4 * rowId, count * 4);
+ }
+
+ @Override
+ public final void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex) {
--- End diff --
The Endianness Platform.getInt() use is platform dependent, when can we use this API?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12635][SQL] Add ColumnarBatch, an in me...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/10628#discussion_r49044395
--- Diff: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/ColumnVector.java ---
@@ -0,0 +1,181 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.vectorized;
+
+import org.apache.spark.sql.types.DataType;
+
+/**
+ * This class represents a column of values and provides the main APIs to access the data
+ * values. It supports all the types and contains get/put APIs as well as their batched versions.
+ * The batched versions are preferable whenever possible.
+ *
+ * Most of the APIs take the rowId as a parameter. This is the local 0-based row id for values
+ * in the current RowBatch.
+ *
+ * A ColumnVector should be considered immutable once originally created. In other words, it is not
+ * valid to call put APIs after reads until reset() is called.
+ */
+public abstract class ColumnVector {
+ /**
+ * Allocates a column with each element of size `width` either on or off heap.
+ */
+ public static ColumnVector allocate(int capacity, DataType type, boolean offHeap) {
+ if (offHeap) {
+ return new OffHeapColumnVector(capacity, type);
+ } else {
+ return new OnHeapColumnVector(capacity, type);
+ }
+ }
+
+ public final DataType dataType() { return type; }
+
+ /**
+ * Resets this column for writing. The currently stored values are no longer accessible.
+ */
+ public void reset() {
+ numNulls = 0;
+ if (anyNullsSet) {
+ putNotNulls(0, capacity);
+ anyNullsSet = false;
+ }
+ }
+
+ /**
+ * Cleans up memory for this column. The column is not usable after this.
+ * TODO: this should probably have ref-counted semantics.
+ */
+ public abstract void close();
+
+ /**
+ * Returns the number of nulls in this column.
+ */
+ public final int numNulls() { return numNulls; }
+
+ /**
+ * Returns true if any of the nulls indicator are set for this column. This can be used
+ * as an optimization to prevent setting nulls.
+ */
+ public final boolean anyNullsSet() { return anyNullsSet; }
+
+ /**
+ * Returns the off heap ptr for the arrays backing the NULLs and values buffer. Only valid
+ * to call for off heap columns.
+ */
+ public abstract long nullsNativeAddress();
+ public abstract long valuesNativeAddress();
+
+ /**
+ * Sets the value at rowId to null/not null.
+ */
+ public abstract void putNotNull(int rowId);
+ public abstract void putNull(int rowId);
+
+ /**
+ * Sets the values from [rowId, rowId + count) to null/not null.
+ */
+ public abstract void putNulls(int rowId, int count);
+ public abstract void putNotNulls(int rowId, int count);
+
+ /**
+ * Returns whether the value at rowId is NULL.
+ */
+ public abstract boolean getIsNull(int rowId);
+
+ /**
+ * Sets the value at rowId to `value`.
+ */
+ public abstract void putInt(int rowId, int value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to value.
+ */
+ public abstract void putInts(int rowId, int count, int value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ */
+ public abstract void putInts(int rowId, int count, int[] src, int srcIndex);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * The data in src must be 4-byte little endian ints.
+ */
+ public abstract void putIntsLittleEndian(int rowId, int count, byte[] src, int srcIndex);
+
+ /**
+ * Returns the integer for rowId.
+ */
+ public abstract int getInt(int rowId);
+
+ /**
+ * Sets the value at rowId to `value`.
+ */
+ public abstract void putDouble(int rowId, double value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to value.
+ */
+ public abstract void putDoubles(int rowId, int count, double value);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * src should contain `count` doubles written as ieee format.
+ */
+ public abstract void putDoubles(int rowId, int count, double[] src, int srcIndex);
+
+ /**
+ * Sets values from [rowId, rowId + count) to [src + srcIndex, src + srcIndex + count)
+ * The data in src must be ieee formated doubles.
+ */
+ public abstract void putDoubles(int rowId, int count, byte[] src, int srcIndex);
+
+ /**
+ * Returns the double for rowId.
+ */
+ public abstract double getDouble(int rowId);
+
+ /**
+ * Maximum number of rows that can be stored in this column.
+ */
+ protected final int capacity;
+
+ /**
+ * Byte width fo this column.
+ */
+ protected final int width;
+
+ /**
+ * Number of nulls in this column.
+ */
+ protected int numNulls;
+
+ /**
+ * True if there is at least one NULL byte set.
+ */
+ protected boolean anyNullsSet;
--- End diff --
Is this equal to `numNulls == 0`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org