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 2016/08/20 14:03:44 UTC
[GitHub] spark pull request #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-...
GitHub user pwoody opened a pull request:
https://github.com/apache/spark/pull/14733
[SPARK-17170] [SQL] InMemoryTableScanExec driver-side partition pruning
## What changes were proposed in this pull request?
After caching data, we have statistics that enable us to eagerly prune entire partitions before launching a query. This modifies the InMemoryTableScanExec to prune partitions before launching the tasks.
## How was this patch tested?
Existing test suite with slight modification to scan over the data once as setup.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/pwoody/spark feature/inMemoryPartitionPruning
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/14733.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 #14733
----
commit 3ef63bd5095e9a42b0813e17b7d15fcb5603d8cc
Author: Patrick Woody <pw...@palantir.com>
Date: 2016-08-19T10:28:13Z
[SPARK-17170] [SQL] InMemoryTableScanExec driver-side partition pruning
----
---
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14733
**[Test build #64659 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64659/consoleFull)** for PR 14733 at commit [`7bf5bb9`](https://github.com/apache/spark/commit/7bf5bb9bacde297abf972a6ae3ad80ba2b6d65fa).
* This patch **fails Spark unit 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 #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14733#discussion_r77257908
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala ---
@@ -125,12 +129,37 @@ case class InMemoryTableScanExec(
val schema = relation.partitionStatistics.schema
val schemaIndex = schema.zipWithIndex
val relOutput: AttributeSeq = relation.output
- val buffers = relation.cachedColumnBuffers
+ val partitionFilter = newPredicate(
+ partitionFilters.reduceOption(And).getOrElse(Literal(true)),
+ schema)
+
+ val buffers = if (inMemoryPartitionPruningEnabled && !relation.batchStats.value.isEmpty) {
+ val validPartitions = relation.batchStats.value.asScala
+ .filter(batchStat => partitionFilter(batchStat._2))
+ .map(_._1)
+ .distinct
+ if (validPartitions.isEmpty) {
+ sparkContext.emptyRDD[CachedBatch]
+ } else {
+ PartitionPruningRDD.create[CachedBatch](relation.cachedColumnBuffers,
+ index => {
+ if (validPartitions.contains(index)) {
+ true
+ } else {
+ logInfo(s"Skipping partition $index because all cached batches will be pruned")
--- End diff --
+1 on logging at debug.
---
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 #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14733#discussion_r77259926
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala ---
@@ -98,9 +100,14 @@ case class InMemoryRelation(
buildBuffers()
}
- def recache(): Unit = {
- _cachedColumnBuffers.unpersist()
+ def unpersist(blocking: Boolean = true): Unit = {
+ batchStats.reset()
--- End diff --
To clarify, is this because the content of the batches might change after recomputation in such a way that the use of these batch stats for whole partition pruning would be invalid? It's my understanding that the _set_ of values in each RDD partition will be the same although their order within that partition may change unless a sort is performed (this is the case for reduce tasks due to interleaving of fetched map output, for example).
Given this, it seems like the correctness case that we'd have to worry about is a situation where the old batch stats would have pruned a partition but that pruning decision is invalid with the new stats. But I'm not sure how that can be the case given that pruning decisions seem to be based on conditions defined over the maximum or minimum values of columns and we're effectively constructing partition-wide stats by `AND`-ing conditions over the per-batch stats.
Basically, I think that I see the motivation for this but I don't have an immediate counterexample to show how things would break if we omitted 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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14733
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 #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-...
Posted by ash211 <gi...@git.apache.org>.
Github user ash211 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14733#discussion_r76714035
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala ---
@@ -125,12 +129,37 @@ case class InMemoryTableScanExec(
val schema = relation.partitionStatistics.schema
val schemaIndex = schema.zipWithIndex
val relOutput: AttributeSeq = relation.output
- val buffers = relation.cachedColumnBuffers
+ val partitionFilter = newPredicate(
+ partitionFilters.reduceOption(And).getOrElse(Literal(true)),
+ schema)
+
+ val buffers = if (inMemoryPartitionPruningEnabled && !relation.batchStats.value.isEmpty) {
+ val validPartitions = relation.batchStats.value.asScala
+ .filter(batchStat => partitionFilter(batchStat._2))
+ .map(_._1)
+ .distinct
+ if (validPartitions.isEmpty) {
+ sparkContext.emptyRDD[CachedBatch]
+ } else {
+ PartitionPruningRDD.create[CachedBatch](relation.cachedColumnBuffers,
+ index => {
+ if (validPartitions.contains(index)) {
+ true
+ } else {
+ logInfo(s"Skipping partition $index because all cached batches will be pruned")
--- End diff --
Log at debug?
---
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on the issue:
https://github.com/apache/spark/pull/14733
Closing stale PR
---
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14733
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14733
**[Test build #64611 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64611/consoleFull)** for PR 14733 at commit [`1631973`](https://github.com/apache/spark/commit/163197394a0976aae622721b9943328381667412).
---
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 #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-...
Posted by robert3005 <gi...@git.apache.org>.
Github user robert3005 commented on a diff in the pull request:
https://github.com/apache/spark/pull/14733#discussion_r75597217
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala ---
@@ -125,12 +129,37 @@ case class InMemoryTableScanExec(
val schema = relation.partitionStatistics.schema
val schemaIndex = schema.zipWithIndex
val relOutput: AttributeSeq = relation.output
- val buffers = relation.cachedColumnBuffers
+ val partitionFilter = newPredicate(
+ partitionFilters.reduceOption(And).getOrElse(Literal(true)),
+ schema)
+
+ val buffers = if (inMemoryPartitionPruningEnabled && !relation.batchStats.value.isEmpty) {
+ val validPartitions = relation.batchStats.value.asScala
+ .filter(batchStat => partitionFilter(batchStat._2))
+ .map(_._1)
+ .distinct
+ if (validPartitions.isEmpty) {
+ new EmptyRDD[CachedBatch](sparkContext)
+ } else {
+ new PartitionPruningRDD[CachedBatch](relation.cachedColumnBuffers,
--- End diff --
There's PartitionPruningRDD.create which will make this slightly cleaner
---
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14733
**[Test build #64611 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64611/consoleFull)** for PR 14733 at commit [`1631973`](https://github.com/apache/spark/commit/163197394a0976aae622721b9943328381667412).
* This patch **fails Spark unit 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 #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody closed the pull request at:
https://github.com/apache/spark/pull/14733
---
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14733
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64611/
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the issue:
https://github.com/apache/spark/pull/14733
I believe that the test failure is happening because `InMemoryTableScanExec`'s `outputPartitioning` says that it has `n` partitions yet the produced RDD may yield fewer than `n` partitions:
```
// The cached version does not change the outputPartitioning of the original SparkPlan.
override def outputPartitioning: Partitioning = relation.child.outputPartitioning
```
I think that this is breaking `SortMergeJoinExec` because it's assuming that both sides of the join have the same number of partitions:
```
left.execute().zipPartitions(right.execute()) { (leftIter, rightIter) =>
val boundCondition: (InternalRow) => Boolean = {
condition.map { cond =>
newPredicate(cond, left.output ++ right.output)
}.getOrElse {
(r: InternalRow) => true
}
}
```
One possible fix would be to have `InMemoryTableScanExec.outputPartitioning` copy the parent partitioning and just adjust the number of partitions, but doing that will possibly introduce performance regressions for certain types of joins joins: for example, in hash joins: even though both sides of a hash join may be partitioned according to the same hash function and modulus Spark won't realize this because the declared number of partitions are different. In other words, you might get lucky and wind up joining two co-partitioned cached datasets and then have the co-partitioning get de-optimized because the partition pruning alters the partitioning.
It might be possible to work around this via a custom, SQL-specific version of ZippedPartitionsRDD which understands how to deal with missing partitions and zips according to partition ids (dealing with gaps), but this will be tricky to get right for outer joins (where you still need to produce output for partitions which match to pruned ones).
---
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14733
**[Test build #64654 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64654/consoleFull)** for PR 14733 at commit [`610d85d`](https://github.com/apache/spark/commit/610d85d3783bdd05b4331af1ed14414e2cd2a422).
---
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 #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14733#discussion_r76717155
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala ---
@@ -63,8 +63,8 @@ case class InMemoryRelation(
@transient child: SparkPlan,
tableName: Option[String])(
@transient var _cachedColumnBuffers: RDD[CachedBatch] = null,
- val batchStats: CollectionAccumulator[InternalRow] =
- child.sqlContext.sparkContext.collectionAccumulator[InternalRow])
+ val batchStats: CollectionAccumulator[(Int, InternalRow)] =
--- End diff --
Maybe add a line comment to explain that `Int` is a partition id?
---
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14733
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14733
**[Test build #64659 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64659/consoleFull)** for PR 14733 at commit [`7bf5bb9`](https://github.com/apache/spark/commit/7bf5bb9bacde297abf972a6ae3ad80ba2b6d65fa).
---
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 #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/14733#discussion_r77308047
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala ---
@@ -142,13 +171,16 @@ case class InMemoryTableScanExec(
val cachedBatchesToScan =
if (inMemoryPartitionPruningEnabled) {
cachedBatchIterator.filter { cachedBatch =>
+ val partitionFilter = newPredicate(
--- End diff --
It was to avoid the call if inMemoryPartitionPruning wasn't enabled. This reasoning is kind of dumb though given that it is the default and if disabled you will pay extra cost elsewhere. I'll move it back.
---
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 #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/14733#discussion_r75672675
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala ---
@@ -125,12 +129,37 @@ case class InMemoryTableScanExec(
val schema = relation.partitionStatistics.schema
val schemaIndex = schema.zipWithIndex
val relOutput: AttributeSeq = relation.output
- val buffers = relation.cachedColumnBuffers
+ val partitionFilter = newPredicate(
+ partitionFilters.reduceOption(And).getOrElse(Literal(true)),
+ schema)
+
+ val buffers = if (inMemoryPartitionPruningEnabled && !relation.batchStats.value.isEmpty) {
+ val validPartitions = relation.batchStats.value.asScala
+ .filter(batchStat => partitionFilter(batchStat._2))
+ .map(_._1)
+ .distinct
+ if (validPartitions.isEmpty) {
+ new EmptyRDD[CachedBatch](sparkContext)
+ } else {
+ new PartitionPruningRDD[CachedBatch](relation.cachedColumnBuffers,
--- End diff --
Cool yeah used this (and removed the EmptyRDD constructor as well). I'd prefer to keep the logging in the function though.
---
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the issue:
https://github.com/apache/spark/pull/14733
This looks like a legitimate test failure:
```
[info] - A cached table preserves the partitioning and ordering of its cached SparkPlan *** FAILED *** (1 second, 305 milliseconds)
[info] Exception thrown while executing query:
[info] == Parsed Logical Plan ==
[info] 'Project [*]
[info] +- 'Join Inner, ('t1.key = 't2.a)
[info] :- 'UnresolvedRelation `t1`, t1
[info] +- 'UnresolvedRelation `t2`, t2
[info]
[info] == Analyzed Logical Plan ==
[info] key: int, value: string, a: int, b: int
[info] Project [key#17868, value#17869, a#19094, b#19095]
[info] +- Join Inner, (key#17868 = a#19094)
[info] :- SubqueryAlias t1, `t1`
[info] : +- RepartitionByExpression [key#17868], 5
[info] : +- SerializeFromObject [assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData, true], top level non-flat input object).key AS key#17868, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData, true], top level non-flat input object).value, true) AS value#17869]
[info] : +- ExternalRDD [obj#17867]
[info] +- SubqueryAlias t2, `t2`
[info] +- RepartitionByExpression [a#19094], 5
[info] +- SerializeFromObject [assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData2, true], top level non-flat input object).a AS a#19094, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData2, true], top level non-flat input object).b AS b#19095]
[info] +- ExternalRDD [obj#19093]
[info]
[info] == Optimized Logical Plan ==
[info] Join Inner, (key#17868 = a#19094)
[info] :- Filter isnotnull(key#17868)
[info] : +- InMemoryRelation [key#17868, value#17869], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas), t1
[info] : +- Exchange hashpartitioning(key#17868, 5)
[info] : +- *SerializeFromObject [assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData, true], top level non-flat input object).key AS key#17868, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData, true], top level non-flat input object).value, true) AS value#17869]
[info] : +- Scan ExternalRDDScan[obj#17867]
[info] +- Filter isnotnull(a#19094)
[info] +- InMemoryRelation [a#19094, b#19095], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas), t2
[info] +- Exchange hashpartitioning(a#19094, 5)
[info] +- *SerializeFromObject [assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData2, true], top level non-flat input object).a AS a#19094, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData2, true], top level non-flat input object).b AS b#19095]
[info] +- Scan ExternalRDDScan[obj#19093]
[info]
[info] == Physical Plan ==
[info] *SortMergeJoin [key#17868], [a#19094], Inner
[info] :- *Sort [key#17868 ASC], false, 0
[info] : +- *Filter isnotnull(key#17868)
[info] : +- InMemoryTableScan [key#17868, value#17869], [isnotnull(key#17868)]
[info] : +- InMemoryRelation [key#17868, value#17869], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas), t1
[info] : +- Exchange hashpartitioning(key#17868, 5)
[info] : +- *SerializeFromObject [assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData, true], top level non-flat input object).key AS key#17868, staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData, true], top level non-flat input object).value, true) AS value#17869]
[info] : +- Scan ExternalRDDScan[obj#17867]
[info] +- *Sort [a#19094 ASC], false, 0
[info] +- *Filter isnotnull(a#19094)
[info] +- InMemoryTableScan [a#19094, b#19095], [isnotnull(a#19094)]
[info] +- InMemoryRelation [a#19094, b#19095], true, 10000, StorageLevel(disk, memory, deserialized, 1 replicas), t2
[info] +- Exchange hashpartitioning(a#19094, 5)
[info] +- *SerializeFromObject [assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData2, true], top level non-flat input object).a AS a#19094, assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData2, true], top level non-flat input object).b AS b#19095]
[info] +- Scan ExternalRDDScan[obj#19093]
[info] == Exception ==
[info] java.lang.IllegalArgumentException: Can't zip RDDs with unequal numbers of partitions: List(5, 3)
[info] java.lang.IllegalArgumentException: Can't zip RDDs with unequal numbers of partitions: List(5, 3)
[info] at org.apache.spark.rdd.ZippedPartitionsBaseRDD.getPartitions(ZippedPartitionsRDD.scala:57)
[info] at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:248)
[info] at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:246)
[info] at scala.Option.getOrElse(Option.scala:121)
[info] at org.apache.spark.rdd.RDD.partitions(RDD.scala:246)
[info] at org.apache.spark.rdd.MapPartitionsRDD.getPartitions(MapPartitionsRDD.scala:35)
[info] at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:248)
[info] at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:246)
[info] at scala.Option.getOrElse(Option.scala:121)
[info] at org.apache.spark.rdd.RDD.partitions(RDD.scala:246)
[info] at org.apache.spark.SparkContext.runJob(SparkContext.scala:1924)
[info] at org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:912)
[info] at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
[info] at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
[info] at org.apache.spark.rdd.RDD.withScope(RDD.scala:358)
[info] at org.apache.spark.rdd.RDD.collect(RDD.scala:911)
[info] at org.apache.spark.sql.execution.SparkPlan.executeCollect(SparkPlan.scala:276)
[info] at org.apache.spark.sql.Dataset$$anonfun$org$apache$spark$sql$Dataset$$execute$1$1.apply(Dataset.scala:2226)
[info] at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:57)
[info] at org.apache.spark.sql.Dataset.withNewExecutionId(Dataset.scala:2576)
[info] at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$execute$1(Dataset.scala:2225)
[info] at org.apache.spark.sql.Dataset$$anonfun$org$apache$spark$sql$Dataset$$collect$1.apply(Dataset.scala:2230)
[info] at org.apache.spark.sql.Dataset$$anonfun$org$apache$spark$sql$Dataset$$collect$1.apply(Dataset.scala:2230)
[info] at org.apache.spark.sql.Dataset.withCallback(Dataset.scala:2589)
[info] at org.apache.spark.sql.Dataset.org$apache$spark$sql$Dataset$$collect(Dataset.scala:2230)
[info] at org.apache.spark.sql.Dataset.collect(Dataset.scala:2206)
[info] at org.apache.spark.sql.QueryTest$.checkAnswer(QueryTest.scala:389)
[info] at org.apache.spark.sql.QueryTest.checkAnswer(QueryTest.scala:175)
[info] at org.apache.spark.sql.QueryTest.checkAnswer(QueryTest.scala:186)
[info] at org.apache.spark.sql.CachedTableSuite$$anonfun$25$$anonfun$apply$mcV$sp$10$$anonfun$apply$mcVI$sp$1.apply$mcV$sp(CachedTableSuite.scala:424)
[info] at org.apache.spark.sql.test.SQLTestUtils$class.withTempView(SQLTestUtils.scala:155)
```
---
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 #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14733#discussion_r76716911
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala ---
@@ -106,7 +106,7 @@ case class InMemoryRelation(
private def buildBuffers(): Unit = {
val output = child.output
- val cached = child.execute().mapPartitionsInternal { rowIterator =>
+ val cached = child.execute().mapPartitionsWithIndex { (i, rowIterator) =>
--- End diff --
Here, I'd prefer to use `TaskContext.get.getPartitionId` to get the partition id. The problem with this change is that it's re-introducing closure cleaning overhead, which `mapPartitionsInternal` avoids.
---
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 #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/14733#discussion_r76775733
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala ---
@@ -125,12 +129,37 @@ case class InMemoryTableScanExec(
val schema = relation.partitionStatistics.schema
val schemaIndex = schema.zipWithIndex
val relOutput: AttributeSeq = relation.output
- val buffers = relation.cachedColumnBuffers
+ val partitionFilter = newPredicate(
+ partitionFilters.reduceOption(And).getOrElse(Literal(true)),
+ schema)
+
+ val buffers = if (inMemoryPartitionPruningEnabled && !relation.batchStats.value.isEmpty) {
+ val validPartitions = relation.batchStats.value.asScala
+ .filter(batchStat => partitionFilter(batchStat._2))
+ .map(_._1)
+ .distinct
+ if (validPartitions.isEmpty) {
+ sparkContext.emptyRDD[CachedBatch]
+ } else {
+ PartitionPruningRDD.create[CachedBatch](relation.cachedColumnBuffers,
+ index => {
+ if (validPartitions.contains(index)) {
+ true
+ } else {
+ logInfo(s"Skipping partition $index because all cached batches will be pruned")
--- End diff --
Current executor-side pruning logging is done at INFO. I have no strong opinion either way, but this can get noisy with many partitions getting pruned.
---
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on the issue:
https://github.com/apache/spark/pull/14733
This looks pretty cool! I'll come back later tonight / tomorrow to test this out and do a more detailed review pass.
---
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14733
**[Test build #64654 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64654/consoleFull)** for PR 14733 at commit [`610d85d`](https://github.com/apache/spark/commit/610d85d3783bdd05b4331af1ed14414e2cd2a422).
* This patch **fails Spark unit 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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14733
Can one of the admins verify this 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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14733
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64654/
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by ash211 <gi...@git.apache.org>.
Github user ash211 commented on the issue:
https://github.com/apache/spark/pull/14733
Jenkins, this is ok to test.
---
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14733
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64659/
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 issue #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-side pa...
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on the issue:
https://github.com/apache/spark/pull/14733
@rxin @davies @dongjoon-hyun mind taking a look?
---
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 #14733: [SPARK-17170] [SQL] InMemoryTableScanExec driver-...
Posted by JoshRosen <gi...@git.apache.org>.
Github user JoshRosen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14733#discussion_r77257778
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryTableScanExec.scala ---
@@ -142,13 +171,16 @@ case class InMemoryTableScanExec(
val cachedBatchesToScan =
if (inMemoryPartitionPruningEnabled) {
cachedBatchIterator.filter { cachedBatch =>
+ val partitionFilter = newPredicate(
--- End diff --
Why did you choose to nest this more deeply inside of the `filter` rather than leaving it where it was in `mapPartitionsInternal`? By moving it here, we'll wind up calling `newPredicate` once per batch rather than once per partition, thereby harming performance.
---
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