You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by sabanas <gi...@git.apache.org> on 2018/03/27 17:51:59 UTC
[GitHub] spark pull request #20915: [SPARK-23803][SQL] Filter prune buckets
GitHub user sabanas opened a pull request:
https://github.com/apache/spark/pull/20915
[SPARK-23803][SQL] Filter prune buckets
## What changes were proposed in this pull request?
support bucket pruning when filtering on a single bucketed column on the following predicates -
EqualTo, EqualNullSafe, In, And/Or predicates
## How was this patch tested?
refactored unit tests to test the above.
based on @gatorsmile work in https://github.com/apache/spark/commit/e3c75c6398b1241500343ff237e9bcf78b5396f9
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/sabanas/spark filter-prune-buckets
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/20915.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 #20915
----
commit 5be4dff693afe5379e4fe0ece081b91b98cc7c9a
Author: Asher Saban <as...@...>
Date: 2018-03-23T23:17:29Z
add bucket pruning functionality
commit 4ab1583d26ab039a8458784d0a93d6aa25077603
Author: Asher Saban <as...@...>
Date: 2018-03-23T23:33:25Z
add composite filters test cases and refactor pruning test
commit 3bb7a2eecf8c2d15e1abc728e8432be9e346e22a
Author: Asher Saban <as...@...>
Date: 2018-03-26T18:52:56Z
remove redundant code and move shared getBucketId to BucketingUtils
commit c45da4b6dd523fd5f2eaa041a5f684ab80db02f4
Author: Asher Saban <as...@...>
Date: 2018-03-27T16:31:34Z
fix variable name and add select buckets count to metadata
commit f0b84bd9231915f30f9643c083b2ab35cbbce472
Author: Asher Saban <as...@...>
Date: 2018-03-27T17:10:25Z
optimize imports
----
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by sabanas <gi...@git.apache.org>.
Github user sabanas commented on the issue:
https://github.com/apache/spark/pull/20915
org.apache.spark.sql.execution.streaming.sources.RateSourceSuite.overflow test failed. it passes on my local dev env and there are several other PRs failing the same test.
possible CI issues? can we run it again?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Filter prune buckets
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/88644/
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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by sabanas <gi...@git.apache.org>.
Github user sabanas commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r178090546
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -50,6 +51,85 @@ import org.apache.spark.sql.execution.SparkPlan
* and add it. Proceed to the next file.
*/
object FileSourceStrategy extends Strategy with Logging {
+
+ // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
+ private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
+ bucketSpec match {
+ case Some(spec) => spec.bucketColumnNames.length == 1 && spec.numBuckets > 1
+ case None => false
+ }
+ }
+
+ private def getExpressionBuckets(expr: Expression,
+ bucketColumnName: String,
+ numBuckets: Int): BitSet = {
+
+ def getMatchedBucketBitSet(attr: Attribute, v: Any): BitSet = {
+ val matchedBuckets = new BitSet(numBuckets)
+ matchedBuckets.set(BucketingUtils.getBucketIdFromValue(attr, numBuckets, v))
+ matchedBuckets
+ }
+
+ expr match {
+ case expressions.EqualTo(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualTo(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.In(a: Attribute, list)
+ if list.forall(_.isInstanceOf[Literal]) && a.name == bucketColumnName =>
+ val valuesSet = list.map(e => e.eval(EmptyRow))
+ valuesSet
+ .map(v => getMatchedBucketBitSet(a, v))
+ .fold(new BitSet(numBuckets))(_ | _)
+ case expressions.IsNull(a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, null)
+ case expressions.And(left, right) =>
+ getExpressionBuckets(left, bucketColumnName, numBuckets) |
--- End diff --
in an AND condition i'd like to get the union of both RHS and LHS BitSets. for example, if table is bucketed on column j:
j == 0 && i == 0 -> will return a bit set matching j's bucket.
i == 0 && k == 0 -> will return union of 2 empty bit sets, meaning an empty bit set.
j == 0 && j == 1 (which is effectively a false condition)-> will return bit set matching both j values. i haven't implemented an optimization for this case (can be done with & condition in case both RHS & LHS are not empty).
can you think of an input that will not work here?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Filter prune buckets
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #88753 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88753/testReport)** for PR 20915 at commit [`697784d`](https://github.com/apache/spark/commit/697784d57642ffe2bda55e518a34265829e4ba2c).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r178137167
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -50,6 +51,85 @@ import org.apache.spark.sql.execution.SparkPlan
* and add it. Proceed to the next file.
*/
object FileSourceStrategy extends Strategy with Logging {
+
+ // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
+ private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
+ bucketSpec match {
+ case Some(spec) => spec.bucketColumnNames.length == 1 && spec.numBuckets > 1
+ case None => false
+ }
+ }
+
+ private def getExpressionBuckets(expr: Expression,
+ bucketColumnName: String,
+ numBuckets: Int): BitSet = {
+
+ def getMatchedBucketBitSet(attr: Attribute, v: Any): BitSet = {
+ val matchedBuckets = new BitSet(numBuckets)
+ matchedBuckets.set(BucketingUtils.getBucketIdFromValue(attr, numBuckets, v))
+ matchedBuckets
+ }
+
+ expr match {
+ case expressions.EqualTo(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualTo(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.In(a: Attribute, list)
+ if list.forall(_.isInstanceOf[Literal]) && a.name == bucketColumnName =>
+ val valuesSet = list.map(e => e.eval(EmptyRow))
+ valuesSet
+ .map(v => getMatchedBucketBitSet(a, v))
+ .fold(new BitSet(numBuckets))(_ | _)
+ case expressions.IsNull(a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, null)
+ case expressions.And(left, right) =>
+ getExpressionBuckets(left, bucketColumnName, numBuckets) |
--- End diff --
Sorry, yeah was mostly curious of the why not have the BitSet modeled as "this bucket might have data" and then doing the natural `&` and `|` operations?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r193420453
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala ---
@@ -90,32 +96,37 @@ abstract class BucketedReadSuite extends QueryTest with SQLTestUtils {
originalDataFrame: DataFrame): Unit = {
// This test verifies parts of the plan. Disable whole stage codegen.
withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false") {
- val strategy = DataSourceStrategy(spark.sessionState.conf)
val bucketedDataFrame = spark.table("bucketed_table").select("i", "j", "k")
val BucketSpec(numBuckets, bucketColumnNames, _) = bucketSpec
// Limit: bucket pruning only works when the bucket column has one and only one column
assert(bucketColumnNames.length == 1)
val bucketColumnIndex = bucketedDataFrame.schema.fieldIndex(bucketColumnNames.head)
val bucketColumn = bucketedDataFrame.schema.toAttributes(bucketColumnIndex)
- val matchedBuckets = new BitSet(numBuckets)
- bucketValues.foreach { value =>
- matchedBuckets.set(strategy.getBucketId(bucketColumn, numBuckets, value))
- }
// Filter could hide the bug in bucket pruning. Thus, skipping all the filters
val plan = bucketedDataFrame.filter(filterCondition).queryExecution.executedPlan
val rdd = plan.find(_.isInstanceOf[DataSourceScanExec])
--- End diff --
nit: not introduced by this PR but this name is wrong, we should probably call it `scanPlan`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #88698 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88698/testReport)** for PR 20915 at commit [`cb36012`](https://github.com/apache/spark/commit/cb36012f8aadc9e6c842e1ff143394dd5de71be5).
* 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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by sabanas <gi...@git.apache.org>.
Github user sabanas commented on the issue:
https://github.com/apache/spark/pull/20915
ping @gatorsmile
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Filter prune buckets
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #88636 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88636/testReport)** for PR 20915 at commit [`f0b84bd`](https://github.com/apache/spark/commit/f0b84bd9231915f30f9643c083b2ab35cbbce472).
* This patch **fails Scala style 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 #20915: [SPARK-23803][SQL] Filter prune buckets
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by sabanas <gi...@git.apache.org>.
Github user sabanas commented on the issue:
https://github.com/apache/spark/pull/20915
@gatorsmile can we please proceed with the review? thanks.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r192953337
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -50,6 +51,84 @@ import org.apache.spark.sql.execution.SparkPlan
* and add it. Proceed to the next file.
*/
object FileSourceStrategy extends Strategy with Logging {
+
+ // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
+ private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
+ bucketSpec match {
+ case Some(spec) => spec.bucketColumnNames.length == 1 && spec.numBuckets > 1
+ case None => false
+ }
+ }
+
+ private def getExpressionBuckets(expr: Expression,
+ bucketColumnName: String,
+ numBuckets: Int): BitSet = {
+
+ def getMatchedBucketBitSet(attr: Attribute, v: Any): BitSet = {
+ val matchedBuckets = new BitSet(numBuckets)
+ matchedBuckets.set(BucketingUtils.getBucketIdFromValue(attr, numBuckets, v))
+ matchedBuckets
+ }
+
+ expr match {
+ case expressions.EqualTo(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
--- End diff --
use `Equality` to match both `EqualTo` and `EqualNullSafe`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r193260674
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala ---
@@ -371,14 +387,27 @@ case class FileSourceScanExec(
val hosts = getBlockHosts(getBlockLocations(f), 0, f.getLen)
PartitionedFile(p.values, f.getPath.toUri.toString, 0, f.getLen, hosts)
}
- }.groupBy { f =>
- BucketingUtils
- .getBucketId(new Path(f.filePath).getName)
- .getOrElse(sys.error(s"Invalid bucket file ${f.filePath}"))
}
+ val prunedBucketed = if (optionalBucketSet.isDefined) {
+ val bucketSet = optionalBucketSet.get
+ bucketed.filter {
+ f => bucketSet.get(
+ BucketingUtils.getBucketId(new Path(f.filePath).getName)
+ .getOrElse(sys.error(s"Invalid bucket file ${f.filePath}")))
+ }
+ } else {
+ bucketed
+ }
+
+ val filesGroupedToBuckets = prunedBucketed.groupBy { f =>
+ BucketingUtils
--- End diff --
can we avoid calculating bucket id from file name twice?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Filter prune buckets
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/88636/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/20915
@sabanas @pwoody Sorry for the late reply. We are migrating Parquet, ORC and the other data sources using Data Source API V2. Thus, we are trying to address all these issues in the new code. @cloud-fan and @gengliangwang are leading the migration efforts now.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Filter prune buckets
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/20915
retest this please.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r192953694
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -50,6 +51,84 @@ import org.apache.spark.sql.execution.SparkPlan
* and add it. Proceed to the next file.
*/
object FileSourceStrategy extends Strategy with Logging {
+
+ // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
+ private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
+ bucketSpec match {
+ case Some(spec) => spec.bucketColumnNames.length == 1 && spec.numBuckets > 1
+ case None => false
+ }
+ }
+
+ private def getExpressionBuckets(expr: Expression,
+ bucketColumnName: String,
+ numBuckets: Int): BitSet = {
+
+ def getMatchedBucketBitSet(attr: Attribute, v: Any): BitSet = {
+ val matchedBuckets = new BitSet(numBuckets)
+ matchedBuckets.set(BucketingUtils.getBucketIdFromValue(attr, numBuckets, v))
+ matchedBuckets
+ }
+
+ expr match {
+ case expressions.EqualTo(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualTo(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.In(a: Attribute, list)
+ if list.forall(_.isInstanceOf[Literal]) && a.name == bucketColumnName =>
+ val valuesSet = list.map(e => e.eval(EmptyRow))
+ valuesSet
+ .map(v => getMatchedBucketBitSet(a, v))
+ .fold(new BitSet(numBuckets))(_ | _)
+ case expressions.IsNull(a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, null)
+ case expressions.And(left, right) =>
+ getExpressionBuckets(left, bucketColumnName, numBuckets) &
+ getExpressionBuckets(right, bucketColumnName, numBuckets)
+ case expressions.Or(left, right) =>
+ getExpressionBuckets(left, bucketColumnName, numBuckets) |
+ getExpressionBuckets(right, bucketColumnName, numBuckets)
+ case _ =>
+ val matchedBuckets = new BitSet(numBuckets)
+ matchedBuckets.setUntil(numBuckets)
+ matchedBuckets
+ }
+ }
+
+ private def getBuckets(normalizedFilters: Seq[Expression],
--- End diff --
nit: code style
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on the issue:
https://github.com/apache/spark/pull/20915
@gatorsmile would appreciate your feedback on this PR. Thanks!
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #88706 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88706/testReport)** for PR 20915 at commit [`8f6bc28`](https://github.com/apache/spark/commit/8f6bc28752a5ef59d8914f4b37e9e47339feb04f).
* 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 #20915: [SPARK-23803][SQL] Filter prune buckets
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #88644 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88644/testReport)** for PR 20915 at commit [`cb36012`](https://github.com/apache/spark/commit/cb36012f8aadc9e6c842e1ff143394dd5de71be5).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on the issue:
https://github.com/apache/spark/pull/20915
Ah, I am happy too that it's added back. Was just simply wondering if I can fix the fix version in SPARK-12850 JIRA. Nothing more then this. Thank you @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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/20915
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/88753/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/20915
thanks, merging to master!
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by HyukjinKwon <gi...@git.apache.org>.
Github user HyukjinKwon commented on the issue:
https://github.com/apache/spark/pull/20915
@cloud-fan, how does it relate to SPARK-23803, SPARK-12850 and SPARK-23507? Was about to take an action to the JIRAs but felt better making sure ahead.
SPARK-12850 was merged in 2.0.0 but reverted by SPARK-14535 in 2.0.0 so it's no problem but SPARK-23803 duplicates SPARK-12850.
and .. you plan to migrate file-based source to datasource v2 which includes refactoring this feature?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
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 pull request #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by sabanas <gi...@git.apache.org>.
Github user sabanas commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r178299616
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -50,6 +51,85 @@ import org.apache.spark.sql.execution.SparkPlan
* and add it. Proceed to the next file.
*/
object FileSourceStrategy extends Strategy with Logging {
+
+ // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
+ private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
+ bucketSpec match {
+ case Some(spec) => spec.bucketColumnNames.length == 1 && spec.numBuckets > 1
+ case None => false
+ }
+ }
+
+ private def getExpressionBuckets(expr: Expression,
+ bucketColumnName: String,
+ numBuckets: Int): BitSet = {
+
+ def getMatchedBucketBitSet(attr: Attribute, v: Any): BitSet = {
+ val matchedBuckets = new BitSet(numBuckets)
+ matchedBuckets.set(BucketingUtils.getBucketIdFromValue(attr, numBuckets, v))
+ matchedBuckets
+ }
+
+ expr match {
+ case expressions.EqualTo(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualTo(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.In(a: Attribute, list)
+ if list.forall(_.isInstanceOf[Literal]) && a.name == bucketColumnName =>
+ val valuesSet = list.map(e => e.eval(EmptyRow))
+ valuesSet
+ .map(v => getMatchedBucketBitSet(a, v))
+ .fold(new BitSet(numBuckets))(_ | _)
+ case expressions.IsNull(a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, null)
+ case expressions.And(left, right) =>
+ getExpressionBuckets(left, bucketColumnName, numBuckets) |
--- End diff --
right, makes sense. i changed it accordingly.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r177973464
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -50,6 +51,80 @@ import org.apache.spark.sql.execution.SparkPlan
* and add it. Proceed to the next file.
*/
object FileSourceStrategy extends Strategy with Logging {
+
+ // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
+ private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
+ bucketSpec match {
+ case Some(spec) => spec.bucketColumnNames.length == 1 && spec.numBuckets > 1
+ case None => false
+ }
+ }
+
+ private def getExpressionBuckets(expr: Expression,
+ bucketColumnName: String,
+ numBuckets: Int): BitSet = {
+
+ def getMatchedBucketBitSet(attr: Attribute, v: Any): BitSet = {
+ val matchedBuckets = new BitSet(numBuckets)
+ matchedBuckets.set(BucketingUtils.getBucketIdFromValue(attr, numBuckets, v))
+ matchedBuckets
+ }
+
+ expr match {
+ case expressions.EqualTo(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualTo(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.In(a: Attribute, list)
+ if list.forall(_.isInstanceOf[Literal]) && a.name == bucketColumnName =>
+ val valuesSet = list.map(e => e.eval(EmptyRow))
+ valuesSet
+ .map(v => getMatchedBucketBitSet(a, v))
+ .fold(new BitSet(numBuckets))(_ | _)
+ case expressions.IsNull(a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, null)
+ case expressions.And(left, right) =>
+ getExpressionBuckets(left, bucketColumnName, numBuckets) |
+ getExpressionBuckets(right, bucketColumnName, numBuckets)
+ case expressions.Or(left, right) =>
+ val leftBuckets = getExpressionBuckets(left, bucketColumnName, numBuckets)
+ val rightBuckets = getExpressionBuckets(right, bucketColumnName, numBuckets)
+
+ // if some expression in OR condition requires all buckets, return an empty BitSet
+ if (leftBuckets.cardinality() == 0 || rightBuckets.cardinality() == 0) {
+ new BitSet(numBuckets)
+ } else {
+ // return a BitSet that includes all required buckets
+ leftBuckets | rightBuckets
+ }
+ case _ => new BitSet(numBuckets)
+ }
+ }
+
+ private def getBuckets(normalizedFilters: Seq[Expression],
+ bucketSpec: BucketSpec): Option[BitSet] = {
+
+ val bucketColumnName = bucketSpec.bucketColumnNames.head
+ val numBuckets = bucketSpec.numBuckets
+
+ val matchedBuckets = normalizedFilters
+ .map(f => getExpressionBuckets(f, bucketColumnName, numBuckets))
+ .fold(new BitSet(numBuckets))(_ | _)
+
+ val numBucketsSelected = if (matchedBuckets.cardinality() != 0) matchedBuckets.cardinality()
--- End diff --
Should match common style in Spark:
```scala
if (matchedBuckets.cardinality() != 0) {
matchedBuckets.cardinality()
} else {
numBuckets
}
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r192953104
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -50,6 +51,84 @@ import org.apache.spark.sql.execution.SparkPlan
* and add it. Proceed to the next file.
*/
object FileSourceStrategy extends Strategy with Logging {
+
+ // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
+ private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
+ bucketSpec match {
+ case Some(spec) => spec.bucketColumnNames.length == 1 && spec.numBuckets > 1
+ case None => false
+ }
+ }
+
+ private def getExpressionBuckets(expr: Expression,
--- End diff --
nit: code style
```
def xxx(
para1: A,
para2: B): XXX
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on the issue:
https://github.com/apache/spark/pull/20915
@gatorsmile any chance you could take a look at this? Let me know if there is someone more appropriate to review, thanks!
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/91500/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/91492/
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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r192953873
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/sources/BucketedReadSuite.scala ---
@@ -83,39 +88,43 @@ abstract class BucketedReadSuite extends QueryTest with SQLTestUtils {
// To verify if the bucket pruning works, this function checks two conditions:
// 1) Check if the pruned buckets (before filtering) are empty.
// 2) Verify the final result is the same as the expected one
- private def checkPrunedAnswers(
- bucketSpec: BucketSpec,
- bucketValues: Seq[Integer],
- filterCondition: Column,
- originalDataFrame: DataFrame): Unit = {
+ private def checkPrunedAnswers(bucketSpec: BucketSpec,
--- End diff --
nit: code style
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #91500 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/91500/testReport)** for PR 20915 at commit [`f949128`](https://github.com/apache/spark/commit/f94912896c1859d078fc52fda867beaecaa68a47).
* 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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/88698/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r192953586
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -50,6 +51,84 @@ import org.apache.spark.sql.execution.SparkPlan
* and add it. Proceed to the next file.
*/
object FileSourceStrategy extends Strategy with Logging {
+
+ // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
+ private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
+ bucketSpec match {
+ case Some(spec) => spec.bucketColumnNames.length == 1 && spec.numBuckets > 1
+ case None => false
+ }
+ }
+
+ private def getExpressionBuckets(expr: Expression,
+ bucketColumnName: String,
+ numBuckets: Int): BitSet = {
+
+ def getMatchedBucketBitSet(attr: Attribute, v: Any): BitSet = {
+ val matchedBuckets = new BitSet(numBuckets)
+ matchedBuckets.set(BucketingUtils.getBucketIdFromValue(attr, numBuckets, v))
+ matchedBuckets
+ }
+
+ expr match {
+ case expressions.EqualTo(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualTo(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.In(a: Attribute, list)
+ if list.forall(_.isInstanceOf[Literal]) && a.name == bucketColumnName =>
+ val valuesSet = list.map(e => e.eval(EmptyRow))
+ valuesSet
+ .map(v => getMatchedBucketBitSet(a, v))
+ .fold(new BitSet(numBuckets))(_ | _)
--- End diff --
can't we create one bit set for all the matched buckets, instead of creating many bit sets and merge them?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by sabanas <gi...@git.apache.org>.
Github user sabanas commented on the issue:
https://github.com/apache/spark/pull/20915
@gatorsmile we would really want to start using this feature in our production stack. could you or anyone else give your feedback on this PR? Thanks!
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Filter prune buckets
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #88644 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88644/testReport)** for PR 20915 at commit [`cb36012`](https://github.com/apache/spark/commit/cb36012f8aadc9e6c842e1ff143394dd5de71be5).
* 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 #20915: [SPARK-23803][SQL] Filter prune buckets
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/91501/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/20915
IIUC the bucket pruning was accidentally removed during refactoring, I'm happy to see it's added back.
The migration must keep all the existing features. I think we have enough tests to guarantee it now.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on the issue:
https://github.com/apache/spark/pull/20915
Ping @gatorsmile
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r178033591
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -50,6 +51,85 @@ import org.apache.spark.sql.execution.SparkPlan
* and add it. Proceed to the next file.
*/
object FileSourceStrategy extends Strategy with Logging {
+
+ // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
+ private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
+ bucketSpec match {
+ case Some(spec) => spec.bucketColumnNames.length == 1 && spec.numBuckets > 1
+ case None => false
+ }
+ }
+
+ private def getExpressionBuckets(expr: Expression,
+ bucketColumnName: String,
+ numBuckets: Int): BitSet = {
+
+ def getMatchedBucketBitSet(attr: Attribute, v: Any): BitSet = {
+ val matchedBuckets = new BitSet(numBuckets)
+ matchedBuckets.set(BucketingUtils.getBucketIdFromValue(attr, numBuckets, v))
+ matchedBuckets
+ }
+
+ expr match {
+ case expressions.EqualTo(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualTo(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.In(a: Attribute, list)
+ if list.forall(_.isInstanceOf[Literal]) && a.name == bucketColumnName =>
+ val valuesSet = list.map(e => e.eval(EmptyRow))
+ valuesSet
+ .map(v => getMatchedBucketBitSet(a, v))
+ .fold(new BitSet(numBuckets))(_ | _)
+ case expressions.IsNull(a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, null)
+ case expressions.And(left, right) =>
+ getExpressionBuckets(left, bucketColumnName, numBuckets) |
--- End diff --
Does this logic work given that empty filters are an empty bitset? Perhaps we can have unmatched filters go to a filled bitset (all buckets *might* match), remove cardinality checks for the ALL match, and swap this to `&`.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #91492 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/91492/testReport)** for PR 20915 at commit [`de9ecb6`](https://github.com/apache/spark/commit/de9ecb64068d1cafa79ce61a2bf1795981a46f46).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #88706 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88706/testReport)** for PR 20915 at commit [`8f6bc28`](https://github.com/apache/spark/commit/8f6bc28752a5ef59d8914f4b37e9e47339feb04f).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/88706/
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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by pwoody <gi...@git.apache.org>.
Github user pwoody commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r178137704
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -50,6 +51,85 @@ import org.apache.spark.sql.execution.SparkPlan
* and add it. Proceed to the next file.
*/
object FileSourceStrategy extends Strategy with Logging {
+
+ // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
+ private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
+ bucketSpec match {
+ case Some(spec) => spec.bucketColumnNames.length == 1 && spec.numBuckets > 1
+ case None => false
+ }
+ }
+
+ private def getExpressionBuckets(expr: Expression,
+ bucketColumnName: String,
+ numBuckets: Int): BitSet = {
+
+ def getMatchedBucketBitSet(attr: Attribute, v: Any): BitSet = {
+ val matchedBuckets = new BitSet(numBuckets)
+ matchedBuckets.set(BucketingUtils.getBucketIdFromValue(attr, numBuckets, v))
+ matchedBuckets
+ }
+
+ expr match {
+ case expressions.EqualTo(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualTo(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.In(a: Attribute, list)
+ if list.forall(_.isInstanceOf[Literal]) && a.name == bucketColumnName =>
+ val valuesSet = list.map(e => e.eval(EmptyRow))
+ valuesSet
+ .map(v => getMatchedBucketBitSet(a, v))
+ .fold(new BitSet(numBuckets))(_ | _)
+ case expressions.IsNull(a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, null)
+ case expressions.And(left, right) =>
+ getExpressionBuckets(left, bucketColumnName, numBuckets) |
+ getExpressionBuckets(right, bucketColumnName, numBuckets)
+ case expressions.Or(left, right) =>
+ val leftBuckets = getExpressionBuckets(left, bucketColumnName, numBuckets)
+ val rightBuckets = getExpressionBuckets(right, bucketColumnName, numBuckets)
+
+ // if some expression in OR condition requires all buckets, return an empty BitSet
+ if (leftBuckets.cardinality() == 0 || rightBuckets.cardinality() == 0) {
+ new BitSet(numBuckets)
+ } else {
+ // return a BitSet that includes all required buckets
+ leftBuckets | rightBuckets
+ }
+ case _ => new BitSet(numBuckets)
+ }
+ }
+
+ private def getBuckets(normalizedFilters: Seq[Expression],
+ bucketSpec: BucketSpec): Option[BitSet] = {
+
+ val bucketColumnName = bucketSpec.bucketColumnNames.head
+ val numBuckets = bucketSpec.numBuckets
+
+ val matchedBuckets = normalizedFilters
--- End diff --
Can we swap this to something like `.reduce(And).map(getExpressionBuckets)`?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #91501 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/91501/testReport)** for PR 20915 at commit [`72712c9`](https://github.com/apache/spark/commit/72712c951338d3acbd1e141717d779ed33b03d82).
* This patch **fails PySpark 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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #88698 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88698/testReport)** for PR 20915 at commit [`cb36012`](https://github.com/apache/spark/commit/cb36012f8aadc9e6c842e1ff143394dd5de71be5).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20915
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 #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r192953456
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -50,6 +51,84 @@ import org.apache.spark.sql.execution.SparkPlan
* and add it. Proceed to the next file.
*/
object FileSourceStrategy extends Strategy with Logging {
+
+ // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
+ private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
+ bucketSpec match {
+ case Some(spec) => spec.bucketColumnNames.length == 1 && spec.numBuckets > 1
+ case None => false
+ }
+ }
+
+ private def getExpressionBuckets(expr: Expression,
+ bucketColumnName: String,
+ numBuckets: Int): BitSet = {
+
+ def getMatchedBucketBitSet(attr: Attribute, v: Any): BitSet = {
+ val matchedBuckets = new BitSet(numBuckets)
+ matchedBuckets.set(BucketingUtils.getBucketIdFromValue(attr, numBuckets, v))
+ matchedBuckets
+ }
+
+ expr match {
+ case expressions.EqualTo(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualTo(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.In(a: Attribute, list)
--- End diff --
we should catch `InSet` as well
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by sabanas <gi...@git.apache.org>.
Github user sabanas commented on the issue:
https://github.com/apache/spark/pull/20915
@cloud-fan thanks for the feedback, I fixed accordingly.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #88753 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88753/testReport)** for PR 20915 at commit [`697784d`](https://github.com/apache/spark/commit/697784d57642ffe2bda55e518a34265829e4ba2c).
* 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 #20915: [SPARK-23803][SQL] Filter prune buckets
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/20915
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 #20915: [SPARK-23803][SQL] Filter prune buckets
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #88636 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/88636/testReport)** for PR 20915 at commit [`f0b84bd`](https://github.com/apache/spark/commit/f0b84bd9231915f30f9643c083b2ab35cbbce472).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r192953858
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -79,6 +158,13 @@ object FileSourceStrategy extends Strategy with Logging {
ExpressionSet(normalizedFilters.filter(_.references.subsetOf(partitionSet)))
logInfo(s"Pruning directories with: ${partitionKeyFilters.mkString(",")}")
+ val bucketSpec: Option[BucketSpec] = fsRelation.bucketSpec
+ val bucketSet = if (shouldPruneBuckets(bucketSpec)) {
+ getBuckets(normalizedFilters, bucketSpec.get)
--- End diff --
nit: maybe better to call it `genBucketSet` instead of `getBuckets`
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by sabanas <gi...@git.apache.org>.
Github user sabanas commented on a diff in the pull request:
https://github.com/apache/spark/pull/20915#discussion_r178299681
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala ---
@@ -50,6 +51,85 @@ import org.apache.spark.sql.execution.SparkPlan
* and add it. Proceed to the next file.
*/
object FileSourceStrategy extends Strategy with Logging {
+
+ // should prune buckets iff num buckets is greater than 1 and there is only one bucket column
+ private def shouldPruneBuckets(bucketSpec: Option[BucketSpec]): Boolean = {
+ bucketSpec match {
+ case Some(spec) => spec.bucketColumnNames.length == 1 && spec.numBuckets > 1
+ case None => false
+ }
+ }
+
+ private def getExpressionBuckets(expr: Expression,
+ bucketColumnName: String,
+ numBuckets: Int): BitSet = {
+
+ def getMatchedBucketBitSet(attr: Attribute, v: Any): BitSet = {
+ val matchedBuckets = new BitSet(numBuckets)
+ matchedBuckets.set(BucketingUtils.getBucketIdFromValue(attr, numBuckets, v))
+ matchedBuckets
+ }
+
+ expr match {
+ case expressions.EqualTo(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualTo(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(a: Attribute, Literal(v, _)) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.EqualNullSafe(Literal(v, _), a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, v)
+ case expressions.In(a: Attribute, list)
+ if list.forall(_.isInstanceOf[Literal]) && a.name == bucketColumnName =>
+ val valuesSet = list.map(e => e.eval(EmptyRow))
+ valuesSet
+ .map(v => getMatchedBucketBitSet(a, v))
+ .fold(new BitSet(numBuckets))(_ | _)
+ case expressions.IsNull(a: Attribute) if a.name == bucketColumnName =>
+ getMatchedBucketBitSet(a, null)
+ case expressions.And(left, right) =>
+ getExpressionBuckets(left, bucketColumnName, numBuckets) |
+ getExpressionBuckets(right, bucketColumnName, numBuckets)
+ case expressions.Or(left, right) =>
+ val leftBuckets = getExpressionBuckets(left, bucketColumnName, numBuckets)
+ val rightBuckets = getExpressionBuckets(right, bucketColumnName, numBuckets)
+
+ // if some expression in OR condition requires all buckets, return an empty BitSet
+ if (leftBuckets.cardinality() == 0 || rightBuckets.cardinality() == 0) {
+ new BitSet(numBuckets)
+ } else {
+ // return a BitSet that includes all required buckets
+ leftBuckets | rightBuckets
+ }
+ case _ => new BitSet(numBuckets)
+ }
+ }
+
+ private def getBuckets(normalizedFilters: Seq[Expression],
+ bucketSpec: BucketSpec): Option[BitSet] = {
+
+ val bucketColumnName = bucketSpec.bucketColumnNames.head
+ val numBuckets = bucketSpec.numBuckets
+
+ val matchedBuckets = normalizedFilters
--- End diff --
done.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #91501 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/91501/testReport)** for PR 20915 at commit [`72712c9`](https://github.com/apache/spark/commit/72712c951338d3acbd1e141717d779ed33b03d82).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #91500 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/91500/testReport)** for PR 20915 at commit [`f949128`](https://github.com/apache/spark/commit/f94912896c1859d078fc52fda867beaecaa68a47).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20915: [SPARK-23803][SQL] Support bucket pruning
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20915
**[Test build #91492 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/91492/testReport)** for PR 20915 at commit [`de9ecb6`](https://github.com/apache/spark/commit/de9ecb64068d1cafa79ce61a2bf1795981a46f46).
* 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