You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by pwoody <gi...@git.apache.org> on 2018/03/29 11:13:22 UTC
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
GitHub user pwoody opened a pull request:
https://github.com/apache/spark/pull/20935
[SPARK-23819][SQL] Fix InMemoryTableScanExec complex type pruning
## What changes were proposed in this pull request?
This PR allows recording of upper/lower bound values in ColumnStats if the data type is orderable.
## How was this patch tested?
Added tests to ColumnStatsSuite and InMemoryColumnarQuerySuite.
Please review http://spark.apache.org/contributing.html before opening a pull request.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/pwoody/spark pw/fixColumnStats
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/20935.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 #20935
----
commit 83e1e5385c8aae52ef14266a6a4c58f4ce9614a9
Author: Patrick Woody <pw...@...>
Date: 2018-03-29T11:06:47Z
SPARK-23819: InMemoryTableScanExec prunes orderable complex types due to out of date ColumnStats
----
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on the issue:
https://github.com/apache/spark/pull/20935
Anything else to be done here?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/88786/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20935
**[Test build #88797 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88797/testReport)** for PR 20935 at commit [`6ea0919`](https://github.com/apache/spark/commit/6ea0919ec0a9dfc6b121c88790fac79aa072bc60).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on the issue:
https://github.com/apache/spark/pull/20935
@cloud-fan @gatorsmile anything @kiszk - any thoughts on this PR?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/88757/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178444386
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnStatsSuite.scala ---
@@ -103,4 +119,109 @@ class ColumnStatsSuite extends SparkFunSuite {
}
}
}
+
+ def testArrayColumnStats(
+ dataType: DataType, orderable: Boolean, initialStatistics: Array[Any]): Unit = {
+ val columnType = ColumnType(dataType)
+
+ test(s"${dataType.typeName}: empty") {
+ val objectStats = new ArrayColumnStats(dataType)
+ objectStats.collectedStatistics.zip(initialStatistics).foreach {
+ case (actual, expected) => assert(actual === expected)
+ }
+ }
+
+ test(s"${dataType.typeName}: non-empty") {
+ import org.apache.spark.sql.execution.columnar.ColumnarTestUtils._
+ val objectStats = new ArrayColumnStats(dataType)
+ val rows = Seq.fill(10)(makeRandomRow(columnType)) ++ Seq.fill(10)(makeNullRow(1))
--- End diff --
Because we don't reuse the unsafe array/row here, we don't actually test on the copying in corresponding column statistics, can we have the test data reusing the unsafe structures to test array and struct column statistics?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/20935
cc @kiszk
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178444706
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +323,75 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
+ protected var upper: T = _
+ protected var lower: T = _
+
+ private val columnType = ColumnType(dataType)
+ private val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
+ }
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
if (!row.isNullAt(ordinal)) {
- val size = columnType.actualSize(row, ordinal)
- sizeInBytes += size
+ sizeInBytes += columnType.actualSize(row, ordinal)
count += 1
+ ordering.foreach { order =>
+ val value = getValue(row, ordinal)
+ if (upper == null || order.gt(value, upper)) upper = copy(value)
+ if (lower == null || order.lt(value, lower)) lower = copy(value)
+ }
} else {
- gatherNullStats
+ gatherNullStats()
}
}
+ def getValue(row: InternalRow, ordinal: Int): T
+
+ def copy(value: T): T
+
+ override def collectedStatistics: Array[Any] =
+ Array[Any](lower, upper, nullCount, count, sizeInBytes)
+}
+
+private[columnar] final class ArrayColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[ArrayData](dataType) {
--- End diff --
Should we be scoping it down? The API for InternalRow gives back ArrayData, so we'd need a cast to do so.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/88797/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178444014
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +323,75 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
+ protected var upper: T = _
+ protected var lower: T = _
+
+ private val columnType = ColumnType(dataType)
+ private val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
--- End diff --
Since this class is only for "orderable", maybe we don't need optional here and `ordering` can just be `Ordering[T]`.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20935
**[Test build #88785 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88785/testReport)** for PR 20935 at commit [`426374b`](https://github.com/apache/spark/commit/426374b6131f65f67f3f023640af37d9a3130585).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178460763
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +324,76 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
+ protected var upper: T = _
+ protected var lower: T = _
+
+ private val columnType = ColumnType(dataType)
+ private val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
+ }
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
if (!row.isNullAt(ordinal)) {
- val size = columnType.actualSize(row, ordinal)
- sizeInBytes += size
+ sizeInBytes += columnType.actualSize(row, ordinal)
count += 1
+ ordering.foreach { order =>
+ val value = getValue(row, ordinal)
+ if (upper == null || order.gt(value, upper)) upper = copy(value)
+ if (lower == null || order.lt(value, lower)) lower = copy(value)
+ }
} else {
- gatherNullStats
+ gatherNullStats()
}
}
+ def getValue(row: InternalRow, ordinal: Int): T
+
+ def copy(value: T): T
+
+ override def collectedStatistics: Array[Any] =
+ Array[Any](lower, upper, nullCount, count, sizeInBytes)
+}
+
+private[columnar] final class ArrayColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[UnsafeArrayData](dataType) {
+ override def getValue(row: InternalRow, ordinal: Int): UnsafeArrayData =
+ row.getArray(ordinal).asInstanceOf[UnsafeArrayData]
+
+ override def copy(value: UnsafeArrayData): UnsafeArrayData = value.copy()
+}
+
+private[columnar] final class StructColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[UnsafeRow](dataType) {
+ private val numFields = dataType.asInstanceOf[StructType].fields.length
+
+ override def getValue(row: InternalRow, ordinal: Int): UnsafeRow =
+ row.getStruct(ordinal, numFields).asInstanceOf[UnsafeRow]
+
+ override def copy(value: UnsafeRow): UnsafeRow = value.copy()
+}
+
+private[columnar] final class MapColumnStats(dataType: DataType) extends ColumnStats {
--- End diff --
Now that you mention it - we can just have it use it now since it will always fall through to the unorderable case. Everything will just work when we make it orderable w/o code change here.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178442839
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnStatsSuite.scala ---
@@ -18,18 +18,35 @@
package org.apache.spark.sql.execution.columnar
import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.util.TypeUtils
import org.apache.spark.sql.types._
class ColumnStatsSuite extends SparkFunSuite {
- testColumnStats(classOf[BooleanColumnStats], BOOLEAN, Array(true, false, 0))
- testColumnStats(classOf[ByteColumnStats], BYTE, Array(Byte.MaxValue, Byte.MinValue, 0))
- testColumnStats(classOf[ShortColumnStats], SHORT, Array(Short.MaxValue, Short.MinValue, 0))
- testColumnStats(classOf[IntColumnStats], INT, Array(Int.MaxValue, Int.MinValue, 0))
- testColumnStats(classOf[LongColumnStats], LONG, Array(Long.MaxValue, Long.MinValue, 0))
- testColumnStats(classOf[FloatColumnStats], FLOAT, Array(Float.MaxValue, Float.MinValue, 0))
- testColumnStats(classOf[DoubleColumnStats], DOUBLE, Array(Double.MaxValue, Double.MinValue, 0))
- testColumnStats(classOf[StringColumnStats], STRING, Array(null, null, 0))
- testDecimalColumnStats(Array(null, null, 0))
+ testColumnStats(classOf[BooleanColumnStats], BOOLEAN, Array(true, false, 0, 0, 0))
--- End diff --
The column statistics have 5 fields in their array, so the zip comparison on the initial stats will drop the final two.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20935
**[Test build #88785 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88785/testReport)** for PR 20935 at commit [`426374b`](https://github.com/apache/spark/commit/426374b6131f65f67f3f023640af37d9a3130585).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178442892
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnarTestUtils.scala ---
@@ -54,12 +54,22 @@ object ColumnarTestUtils {
case COMPACT_DECIMAL(precision, scale) => Decimal(Random.nextLong() % 100, precision, scale)
case LARGE_DECIMAL(precision, scale) => Decimal(Random.nextLong(), precision, scale)
case STRUCT(_) =>
- new GenericInternalRow(Array[Any](UTF8String.fromString(Random.nextString(10))))
+ val schema = StructType(Array(StructField("test", StringType)))
+ val converter = UnsafeProjection.create(schema)
+ converter(InternalRow(Array(UTF8String.fromString(Random.nextString(10))): _*))
case ARRAY(_) =>
- new GenericArrayData(Array[Any](Random.nextInt(), Random.nextInt()))
+ UnsafeArrayData.fromPrimitiveArray(Array(Random.nextInt(), Random.nextInt()))
case MAP(_) =>
- ArrayBasedMapData(
- Map(Random.nextInt() -> UTF8String.fromString(Random.nextString(Random.nextInt(32)))))
+ val unsafeConverter =
+ UnsafeProjection.create(Array[DataType](MapType(IntegerType, StringType)))
+ val row = new GenericInternalRow(1)
+ def toUnsafeMap(map: ArrayBasedMapData): UnsafeMapData = {
+ row.update(0, map)
+ val unsafeRow = unsafeConverter.apply(row)
+ unsafeRow.getMap(0).copy
+ }
+ toUnsafeMap(ArrayBasedMapData(
+ Map(Random.nextInt() -> UTF8String.fromString(Random.nextString(Random.nextInt(32))))))
--- End diff --
The ColumnType for Maps/Struct/Array all end up casting to their Unsafe structures to get the size for the statistics, so the test data will need to reflect that as well.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178446476
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +324,76 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
+ protected var upper: T = _
+ protected var lower: T = _
+
+ private val columnType = ColumnType(dataType)
+ private val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
+ }
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
if (!row.isNullAt(ordinal)) {
- val size = columnType.actualSize(row, ordinal)
- sizeInBytes += size
+ sizeInBytes += columnType.actualSize(row, ordinal)
count += 1
+ ordering.foreach { order =>
+ val value = getValue(row, ordinal)
+ if (upper == null || order.gt(value, upper)) upper = copy(value)
+ if (lower == null || order.lt(value, lower)) lower = copy(value)
+ }
} else {
- gatherNullStats
+ gatherNullStats()
}
}
+ def getValue(row: InternalRow, ordinal: Int): T
+
+ def copy(value: T): T
+
+ override def collectedStatistics: Array[Any] =
+ Array[Any](lower, upper, nullCount, count, sizeInBytes)
+}
+
+private[columnar] final class ArrayColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[UnsafeArrayData](dataType) {
+ override def getValue(row: InternalRow, ordinal: Int): UnsafeArrayData =
+ row.getArray(ordinal).asInstanceOf[UnsafeArrayData]
+
+ override def copy(value: UnsafeArrayData): UnsafeArrayData = value.copy()
+}
+
+private[columnar] final class StructColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[UnsafeRow](dataType) {
+ private val numFields = dataType.asInstanceOf[StructType].fields.length
+
+ override def getValue(row: InternalRow, ordinal: Int): UnsafeRow =
+ row.getStruct(ordinal, numFields).asInstanceOf[UnsafeRow]
+
+ override def copy(value: UnsafeRow): UnsafeRow = value.copy()
+}
+
+private[columnar] final class MapColumnStats(dataType: DataType) extends ColumnStats {
--- End diff --
dataType: DataType -> dataType: MapType?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178444697
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +323,75 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
+ protected var upper: T = _
+ protected var lower: T = _
+
+ private val columnType = ColumnType(dataType)
+ private val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
--- End diff --
This is for DataTypes that *could* be orderable since Arrays and Structs may have children data types that aren't.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/20935
I'd like to review this PR after the parquet nested column pruning is merged.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178444713
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnStatsSuite.scala ---
@@ -103,4 +119,109 @@ class ColumnStatsSuite extends SparkFunSuite {
}
}
}
+
+ def testArrayColumnStats(
+ dataType: DataType, orderable: Boolean, initialStatistics: Array[Any]): Unit = {
+ val columnType = ColumnType(dataType)
+
+ test(s"${dataType.typeName}: empty") {
+ val objectStats = new ArrayColumnStats(dataType)
+ objectStats.collectedStatistics.zip(initialStatistics).foreach {
+ case (actual, expected) => assert(actual === expected)
+ }
+ }
+
+ test(s"${dataType.typeName}: non-empty") {
+ import org.apache.spark.sql.execution.columnar.ColumnarTestUtils._
+ val objectStats = new ArrayColumnStats(dataType)
+ val rows = Seq.fill(10)(makeRandomRow(columnType)) ++ Seq.fill(10)(makeNullRow(1))
--- End diff --
Yep, will do.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/20935
cc @cloud-fan
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178425525
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnarTestUtils.scala ---
@@ -54,12 +54,22 @@ object ColumnarTestUtils {
case COMPACT_DECIMAL(precision, scale) => Decimal(Random.nextLong() % 100, precision, scale)
case LARGE_DECIMAL(precision, scale) => Decimal(Random.nextLong(), precision, scale)
case STRUCT(_) =>
- new GenericInternalRow(Array[Any](UTF8String.fromString(Random.nextString(10))))
+ val schema = StructType(Array(StructField("test", StringType)))
+ val converter = UnsafeProjection.create(schema)
+ converter(InternalRow(Array(UTF8String.fromString(Random.nextString(10))): _*))
case ARRAY(_) =>
- new GenericArrayData(Array[Any](Random.nextInt(), Random.nextInt()))
+ UnsafeArrayData.fromPrimitiveArray(Array(Random.nextInt(), Random.nextInt()))
case MAP(_) =>
- ArrayBasedMapData(
- Map(Random.nextInt() -> UTF8String.fromString(Random.nextString(Random.nextInt(32)))))
+ val unsafeConverter =
+ UnsafeProjection.create(Array[DataType](MapType(IntegerType, StringType)))
+ val row = new GenericInternalRow(1)
+ def toUnsafeMap(map: ArrayBasedMapData): UnsafeMapData = {
+ row.update(0, map)
+ val unsafeRow = unsafeConverter.apply(row)
+ unsafeRow.getMap(0).copy
+ }
+ toUnsafeMap(ArrayBasedMapData(
+ Map(Random.nextInt() -> UTF8String.fromString(Random.nextString(Random.nextInt(32))))))
--- End diff --
Seems above changes to data generation are unnecessary too?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178444711
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +323,75 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
+ protected var upper: T = _
+ protected var lower: T = _
+
+ private val columnType = ColumnType(dataType)
+ private val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
+ }
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
if (!row.isNullAt(ordinal)) {
- val size = columnType.actualSize(row, ordinal)
- sizeInBytes += size
+ sizeInBytes += columnType.actualSize(row, ordinal)
count += 1
+ ordering.foreach { order =>
+ val value = getValue(row, ordinal)
+ if (upper == null || order.gt(value, upper)) upper = copy(value)
+ if (lower == null || order.lt(value, lower)) lower = copy(value)
+ }
} else {
- gatherNullStats
+ gatherNullStats()
}
}
+ def getValue(row: InternalRow, ordinal: Int): T
+
+ def copy(value: T): T
+
+ override def collectedStatistics: Array[Any] =
+ Array[Any](lower, upper, nullCount, count, sizeInBytes)
+}
+
+private[columnar] final class ArrayColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[ArrayData](dataType) {
+ override def getValue(row: InternalRow, ordinal: Int): ArrayData = row.getArray(ordinal)
+
+ override def copy(value: ArrayData): ArrayData = value.copy()
+}
+
+private[columnar] final class StructColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[InternalRow](dataType) {
--- End diff --
Same question as for the ArrayData above.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178444026
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -307,7 +308,7 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
// TODO: this is not right for DecimalType with precision > 18
gatherValueStats(value)
} else {
- gatherNullStats
+ gatherNullStats()
--- End diff --
I don't think the change to `gatherNullStats` is necessary...
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178446463
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +324,76 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
+ protected var upper: T = _
+ protected var lower: T = _
+
+ private val columnType = ColumnType(dataType)
+ private val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
+ }
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
if (!row.isNullAt(ordinal)) {
- val size = columnType.actualSize(row, ordinal)
- sizeInBytes += size
+ sizeInBytes += columnType.actualSize(row, ordinal)
count += 1
+ ordering.foreach { order =>
+ val value = getValue(row, ordinal)
+ if (upper == null || order.gt(value, upper)) upper = copy(value)
+ if (lower == null || order.lt(value, lower)) lower = copy(value)
+ }
} else {
- gatherNullStats
+ gatherNullStats()
}
}
+ def getValue(row: InternalRow, ordinal: Int): T
+
+ def copy(value: T): T
+
+ override def collectedStatistics: Array[Any] =
+ Array[Any](lower, upper, nullCount, count, sizeInBytes)
+}
+
+private[columnar] final class ArrayColumnStats(dataType: DataType)
--- End diff --
dataType: DataType -> dataType: ArrayType?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178443991
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +323,75 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
--- End diff --
`OrderableObjectColumnStats`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178444088
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnStatsSuite.scala ---
@@ -103,4 +119,109 @@ class ColumnStatsSuite extends SparkFunSuite {
}
}
}
+
+ def testArrayColumnStats(
+ dataType: DataType, orderable: Boolean, initialStatistics: Array[Any]): Unit = {
+ val columnType = ColumnType(dataType)
+
+ test(s"${dataType.typeName}: empty") {
+ val objectStats = new ArrayColumnStats(dataType)
+ objectStats.collectedStatistics.zip(initialStatistics).foreach {
+ case (actual, expected) => assert(actual === expected)
+ }
+ }
+
+ test(s"${dataType.typeName}: non-empty") {
+ import org.apache.spark.sql.execution.columnar.ColumnarTestUtils._
+ val objectStats = new ArrayColumnStats(dataType)
+ val rows = Seq.fill(10)(makeRandomRow(columnType)) ++ Seq.fill(10)(makeNullRow(1))
+ rows.foreach(objectStats.gatherStats(_, 0))
+
+ val stats = objectStats.collectedStatistics
+ if (orderable) {
+ val values = rows.take(10).map(_.get(0, columnType.dataType))
+ val ordering = TypeUtils.getInterpretedOrdering(dataType)
+
+ assertResult(values.min(ordering), "Wrong lower bound")(stats(0))
+ assertResult(values.max(ordering), "Wrong upper bound")(stats(1))
+ } else {
+ assertResult(null, "Wrong lower bound")(stats(0))
+ assertResult(null, "Wrong upper bound")(stats(1))
+ }
+ assertResult(10, "Wrong null count")(stats(2))
+ assertResult(20, "Wrong row count")(stats(3))
+ assertResult(stats(4), "Wrong size in bytes") {
+ rows.map { row =>
+ if (row.isNullAt(0)) 4 else columnType.actualSize(row, 0)
+ }.sum
+ }
+ }
+ }
+
+ def testStructColumnStats(
--- End diff --
Can't we merge `testArrayColumnStats`, `testMapColumnStats` and `testStructColumnStats`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20935
**[Test build #88786 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88786/testReport)** for PR 20935 at commit [`1479bde`](https://github.com/apache/spark/commit/1479bded7c6e220f6dad2eb5bff4feb64cbaf991).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178446592
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnStatsSuite.scala ---
@@ -103,4 +140,90 @@ class ColumnStatsSuite extends SparkFunSuite {
}
}
}
+
+ def testOrderableColumnStats[T](
+ dataType: DataType,
+ statsSupplier: () => OrderableSafeColumnStats[T],
+ columnType: ColumnType[T],
+ orderable: Boolean,
+ initialStatistics: Array[Any]): Unit = {
+
+ test(s"${dataType.typeName}, $orderable: empty") {
+ val objectStats = statsSupplier()
+ objectStats.collectedStatistics.zip(initialStatistics).foreach {
+ case (actual, expected) => assert(actual === expected)
+ }
+ }
+
+ test(s"${dataType.typeName}, $orderable: non-empty") {
+ import org.apache.spark.sql.execution.columnar.ColumnarTestUtils._
+ val objectStats = statsSupplier()
+ val rows = Seq.fill(10)(makeRandomRow(columnType)) ++ Seq.fill(10)(makeNullRow(1))
+ rows.foreach(objectStats.gatherStats(_, 0))
+
+ val stats = objectStats.collectedStatistics
+ if (orderable) {
+ val values = rows.take(10).map(_.get(0, columnType.dataType))
+ val ordering = TypeUtils.getInterpretedOrdering(dataType)
+
+ assertResult(values.min(ordering), "Wrong lower bound")(stats(0))
+ assertResult(values.max(ordering), "Wrong upper bound")(stats(1))
+ } else {
+ assertResult(null, "Wrong lower bound")(stats(0))
+ assertResult(null, "Wrong upper bound")(stats(1))
+ }
+ assertResult(10, "Wrong null count")(stats(2))
+ assertResult(20, "Wrong row count")(stats(3))
+ assertResult(stats(4), "Wrong size in bytes") {
+ rows.map { row =>
+ if (row.isNullAt(0)) 4 else columnType.actualSize(row, 0)
+ }.sum
+ }
+ }
+ }
+
+ def testMapColumnStats(dataType: DataType, initialStatistics: Array[Any]): Unit = {
+ val columnType = ColumnType(dataType)
+
+ test(s"${dataType.typeName}: empty") {
+ val objectStats = new MapColumnStats(dataType)
+ objectStats.collectedStatistics.zip(initialStatistics).foreach {
+ case (actual, expected) => assert(actual === expected)
+ }
+ }
+
+ test(s"${dataType.typeName}: non-empty") {
+ import org.apache.spark.sql.execution.columnar.ColumnarTestUtils._
+ val objectStats = new MapColumnStats(dataType)
+ val rows = Seq.fill(10)(makeRandomRow(columnType)) ++ Seq.fill(10)(makeNullRow(1))
+ rows.foreach(objectStats.gatherStats(_, 0))
+
+ val stats = objectStats.collectedStatistics
+ assertResult(null, "Wrong lower bound")(stats(0))
+ assertResult(null, "Wrong upper bound")(stats(1))
+ assertResult(10, "Wrong null count")(stats(2))
+ assertResult(20, "Wrong row count")(stats(3))
+ assertResult(stats(4), "Wrong size in bytes") {
+ rows.map { row =>
+ if (row.isNullAt(0)) 4 else columnType.actualSize(row, 0)
+ }.sum
+ }
+ }
+ }
+
+ test("Reuse UnsafeArrayData for stats") {
--- End diff --
We should also test against UnsafeRow too.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178444321
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +323,75 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
+ protected var upper: T = _
+ protected var lower: T = _
+
+ private val columnType = ColumnType(dataType)
+ private val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
+ }
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
if (!row.isNullAt(ordinal)) {
- val size = columnType.actualSize(row, ordinal)
- sizeInBytes += size
+ sizeInBytes += columnType.actualSize(row, ordinal)
count += 1
+ ordering.foreach { order =>
+ val value = getValue(row, ordinal)
+ if (upper == null || order.gt(value, upper)) upper = copy(value)
+ if (lower == null || order.lt(value, lower)) lower = copy(value)
+ }
} else {
- gatherNullStats
+ gatherNullStats()
}
}
+ def getValue(row: InternalRow, ordinal: Int): T
+
+ def copy(value: T): T
+
+ override def collectedStatistics: Array[Any] =
+ Array[Any](lower, upper, nullCount, count, sizeInBytes)
+}
+
+private[columnar] final class ArrayColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[ArrayData](dataType) {
+ override def getValue(row: InternalRow, ordinal: Int): ArrayData = row.getArray(ordinal)
+
+ override def copy(value: ArrayData): ArrayData = value.copy()
+}
+
+private[columnar] final class StructColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[InternalRow](dataType) {
--- End diff --
InternalRow -> UnsafeRow? Looks like for struct, the column type is specified for `UnsafeRow`.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20935
**[Test build #93060 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93060/testReport)** for PR 20935 at commit [`6ea0919`](https://github.com/apache/spark/commit/6ea0919ec0a9dfc6b121c88790fac79aa072bc60).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20935
**[Test build #93060 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/93060/testReport)** for PR 20935 at commit [`6ea0919`](https://github.com/apache/spark/commit/6ea0919ec0a9dfc6b121c88790fac79aa072bc60).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/88785/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178425464
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnStatsSuite.scala ---
@@ -18,18 +18,35 @@
package org.apache.spark.sql.execution.columnar
import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.expressions.RowOrdering
+import org.apache.spark.sql.catalyst.util.TypeUtils
import org.apache.spark.sql.types._
class ColumnStatsSuite extends SparkFunSuite {
- testColumnStats(classOf[BooleanColumnStats], BOOLEAN, Array(true, false, 0))
- testColumnStats(classOf[ByteColumnStats], BYTE, Array(Byte.MaxValue, Byte.MinValue, 0))
- testColumnStats(classOf[ShortColumnStats], SHORT, Array(Short.MaxValue, Short.MinValue, 0))
- testColumnStats(classOf[IntColumnStats], INT, Array(Int.MaxValue, Int.MinValue, 0))
- testColumnStats(classOf[LongColumnStats], LONG, Array(Long.MaxValue, Long.MinValue, 0))
- testColumnStats(classOf[FloatColumnStats], FLOAT, Array(Float.MaxValue, Float.MinValue, 0))
- testColumnStats(classOf[DoubleColumnStats], DOUBLE, Array(Double.MaxValue, Double.MinValue, 0))
- testColumnStats(classOf[StringColumnStats], STRING, Array(null, null, 0))
- testDecimalColumnStats(Array(null, null, 0))
+ testColumnStats(classOf[BooleanColumnStats], BOOLEAN, Array(true, false, 0, 0, 0))
--- End diff --
Those changes to `testColumnStats` seems unnecessary?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on the issue:
https://github.com/apache/spark/pull/20935
ok to test
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20935
**[Test build #88757 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88757/testReport)** for PR 20935 at commit [`5c95cef`](https://github.com/apache/spark/commit/5c95cef81257185205c7ecebe3353060780b2c52).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178446501
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +324,76 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
+ protected var upper: T = _
+ protected var lower: T = _
+
+ private val columnType = ColumnType(dataType)
+ private val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
+ }
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
if (!row.isNullAt(ordinal)) {
- val size = columnType.actualSize(row, ordinal)
- sizeInBytes += size
+ sizeInBytes += columnType.actualSize(row, ordinal)
count += 1
+ ordering.foreach { order =>
+ val value = getValue(row, ordinal)
+ if (upper == null || order.gt(value, upper)) upper = copy(value)
+ if (lower == null || order.lt(value, lower)) lower = copy(value)
+ }
} else {
- gatherNullStats
+ gatherNullStats()
}
}
+ def getValue(row: InternalRow, ordinal: Int): T
+
+ def copy(value: T): T
+
+ override def collectedStatistics: Array[Any] =
+ Array[Any](lower, upper, nullCount, count, sizeInBytes)
+}
+
+private[columnar] final class ArrayColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[UnsafeArrayData](dataType) {
+ override def getValue(row: InternalRow, ordinal: Int): UnsafeArrayData =
+ row.getArray(ordinal).asInstanceOf[UnsafeArrayData]
+
+ override def copy(value: UnsafeArrayData): UnsafeArrayData = value.copy()
+}
+
+private[columnar] final class StructColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[UnsafeRow](dataType) {
+ private val numFields = dataType.asInstanceOf[StructType].fields.length
+
+ override def getValue(row: InternalRow, ordinal: Int): UnsafeRow =
+ row.getStruct(ordinal, numFields).asInstanceOf[UnsafeRow]
+
+ override def copy(value: UnsafeRow): UnsafeRow = value.copy()
+}
+
+private[columnar] final class MapColumnStats(dataType: DataType) extends ColumnStats {
--- End diff --
Please add a TODO that we need to make this use `OrderableSafeColumnStats` when MapType is orderable.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on the issue:
https://github.com/apache/spark/pull/20935
Ping @cloud-fan @gatorsmile @kiszk
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/93060/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178444724
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -307,7 +308,7 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
// TODO: this is not right for DecimalType with precision > 18
gatherValueStats(value)
} else {
- gatherNullStats
+ gatherNullStats()
--- End diff --
Yeah this was mostly from the scala style guide since this mutates the backing stats. http://docs.scala-lang.org/style/method-invocation.html#arity-0
I don't have a strong opinion though, so happy to swap it back.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178444325
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +323,75 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
+ protected var upper: T = _
+ protected var lower: T = _
+
+ private val columnType = ColumnType(dataType)
+ private val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
+ }
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
if (!row.isNullAt(ordinal)) {
- val size = columnType.actualSize(row, ordinal)
- sizeInBytes += size
+ sizeInBytes += columnType.actualSize(row, ordinal)
count += 1
+ ordering.foreach { order =>
+ val value = getValue(row, ordinal)
+ if (upper == null || order.gt(value, upper)) upper = copy(value)
+ if (lower == null || order.lt(value, lower)) lower = copy(value)
+ }
} else {
- gatherNullStats
+ gatherNullStats()
}
}
+ def getValue(row: InternalRow, ordinal: Int): T
+
+ def copy(value: T): T
+
+ override def collectedStatistics: Array[Any] =
+ Array[Any](lower, upper, nullCount, count, sizeInBytes)
+}
+
+private[columnar] final class ArrayColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[ArrayData](dataType) {
--- End diff --
ArrayData -> UnsafeArrayData?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178442841
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -323,18 +324,31 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
}
private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
+ protected var upper: Any = null
+ protected var lower: Any = null
+
val columnType = ColumnType(dataType)
+ val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) && x != NullType =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
+ }
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
if (!row.isNullAt(ordinal)) {
val size = columnType.actualSize(row, ordinal)
sizeInBytes += size
count += 1
+ ordering.foreach { order =>
+ val value = row.get(ordinal, dataType)
+ if (upper == null || order.gt(value, upper)) upper = value
+ if (lower == null || order.lt(value, lower)) lower = value
--- End diff --
Yes, thanks for catching this.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178446454
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +324,76 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
+ protected var upper: T = _
+ protected var lower: T = _
+
+ private val columnType = ColumnType(dataType)
+ private val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
+ }
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
if (!row.isNullAt(ordinal)) {
- val size = columnType.actualSize(row, ordinal)
- sizeInBytes += size
+ sizeInBytes += columnType.actualSize(row, ordinal)
count += 1
+ ordering.foreach { order =>
+ val value = getValue(row, ordinal)
+ if (upper == null || order.gt(value, upper)) upper = copy(value)
+ if (lower == null || order.lt(value, lower)) lower = copy(value)
+ }
} else {
- gatherNullStats
+ gatherNullStats()
}
}
+ def getValue(row: InternalRow, ordinal: Int): T
+
+ def copy(value: T): T
+
+ override def collectedStatistics: Array[Any] =
+ Array[Any](lower, upper, nullCount, count, sizeInBytes)
+}
+
+private[columnar] final class ArrayColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[UnsafeArrayData](dataType) {
+ override def getValue(row: InternalRow, ordinal: Int): UnsafeArrayData =
+ row.getArray(ordinal).asInstanceOf[UnsafeArrayData]
+
+ override def copy(value: UnsafeArrayData): UnsafeArrayData = value.copy()
+}
+
+private[columnar] final class StructColumnStats(dataType: DataType)
--- End diff --
dataType: DataType -> dataType: StructType?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20935
**[Test build #88786 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88786/testReport)** for PR 20935 at commit [`1479bde`](https://github.com/apache/spark/commit/1479bded7c6e220f6dad2eb5bff4feb64cbaf991).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/20935
ok to test
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178425406
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -323,18 +324,31 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
}
private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
+ protected var upper: Any = null
+ protected var lower: Any = null
+
val columnType = ColumnType(dataType)
+ val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) && x != NullType =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
+ }
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
if (!row.isNullAt(ordinal)) {
val size = columnType.actualSize(row, ordinal)
sizeInBytes += size
count += 1
+ ordering.foreach { order =>
+ val value = row.get(ordinal, dataType)
+ if (upper == null || order.gt(value, upper)) upper = value
+ if (lower == null || order.lt(value, lower)) lower = value
--- End diff --
For unsafe row and array, doesn't we need to copy the value? In the added test this can't be tested because the random rows are all individual instances, however, it can be the same instance of unsafe row or array during query evaluation.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec comp...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20935#discussion_r178992356
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ColumnStats.scala ---
@@ -322,19 +324,76 @@ private[columnar] final class DecimalColumnStats(precision: Int, scale: Int) ext
Array[Any](lower, upper, nullCount, count, sizeInBytes)
}
-private[columnar] final class ObjectColumnStats(dataType: DataType) extends ColumnStats {
- val columnType = ColumnType(dataType)
+private abstract class OrderableSafeColumnStats[T](dataType: DataType) extends ColumnStats {
+ protected var upper: T = _
+ protected var lower: T = _
+
+ private val columnType = ColumnType(dataType)
+ private val ordering = dataType match {
+ case x if RowOrdering.isOrderable(dataType) =>
+ Option(TypeUtils.getInterpretedOrdering(x))
+ case _ => None
+ }
override def gatherStats(row: InternalRow, ordinal: Int): Unit = {
if (!row.isNullAt(ordinal)) {
- val size = columnType.actualSize(row, ordinal)
- sizeInBytes += size
+ sizeInBytes += columnType.actualSize(row, ordinal)
count += 1
+ ordering.foreach { order =>
+ val value = getValue(row, ordinal)
+ if (upper == null || order.gt(value, upper)) upper = copy(value)
+ if (lower == null || order.lt(value, lower)) lower = copy(value)
+ }
} else {
- gatherNullStats
+ gatherNullStats()
}
}
+ def getValue(row: InternalRow, ordinal: Int): T
+
+ def copy(value: T): T
+
+ override def collectedStatistics: Array[Any] =
+ Array[Any](lower, upper, nullCount, count, sizeInBytes)
+}
+
+private[columnar] final class ArrayColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[UnsafeArrayData](dataType) {
+ override def getValue(row: InternalRow, ordinal: Int): UnsafeArrayData =
+ row.getArray(ordinal).asInstanceOf[UnsafeArrayData]
+
+ override def copy(value: UnsafeArrayData): UnsafeArrayData = value.copy()
+}
+
+private[columnar] final class StructColumnStats(dataType: DataType)
+ extends OrderableSafeColumnStats[UnsafeRow](dataType) {
+ private val numFields = dataType.asInstanceOf[StructType].fields.length
+
+ override def getValue(row: InternalRow, ordinal: Int): UnsafeRow =
+ row.getStruct(ordinal, numFields).asInstanceOf[UnsafeRow]
+
+ override def copy(value: UnsafeRow): UnsafeRow = value.copy()
+}
+
+private[columnar] final class MapColumnStats(dataType: DataType) extends ColumnStats {
--- End diff --
Oh, sounds good to me.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20935
**[Test build #88797 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88797/testReport)** for PR 20935 at commit [`6ea0919`](https://github.com/apache/spark/commit/6ea0919ec0a9dfc6b121c88790fac79aa072bc60).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20935
**[Test build #88757 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88757/testReport)** for PR 20935 at commit [`5c95cef`](https://github.com/apache/spark/commit/5c95cef81257185205c7ecebe3353060780b2c52).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20935: [SPARK-23819][SQL] Fix InMemoryTableScanExec complex typ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20935
Can one of the admins verify this patch?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org