Uh oh!
There was an error while loading. Please reload this page.
[SPARK-17091][SQL] Add rule to convert IN predicate to equivalent Parquet filter - #21603
Closed
wangyum wants to merge 10 commits into
Closed
[SPARK-17091][SQL] Add rule to convert IN predicate to equivalent Parquet filter#21603wangyum wants to merge 10 commits into
wangyum wants to merge 10 commits into
Conversation
wangyum
commented
Jun 21, 2018
| case sources.Not(pred) => | ||
| createFilter(schema, pred).map(FilterApi.not) | ||
| case sources.In(name, values) if canMakeFilterOn(name) && values.length < 20 => |
MemberAuthor
There was a problem hiding this comment.
The threshold is 20. Too many values may be OOM, for example:
spark.range(10000000).coalesce(1).write.option("parquet.block.size", 1048576).parquet("/tmp/spark/parquet/SPARK-17091")
valdf= spark.read.parquet("/tmp/spark/parquet/SPARK-17091/")
df.where(s"id in(${Range(1, 10000).mkString(",")})").countException in thread "SIGINT handler" 18/06/21 13:00:54 WARN TaskSetManager: Lost task 7.0 in stage 1.0 (TID 8, localhost, executor driver): java.lang.OutOfMemoryError: Java heap space
at java.util.Arrays.copyOfRange(Arrays.java:3664)
at java.lang.String.<init>(String.java:207)
at java.lang.StringBuilder.toString(StringBuilder.java:407)
at org.apache.parquet.filter2.predicate.Operators$BinaryLogicalFilterPredicate.<init>(Operators.java:263)
at org.apache.parquet.filter2.predicate.Operators$Or.<init>(Operators.java:316)
at org.apache.parquet.filter2.predicate.FilterApi.or(FilterApi.java:261)
at org.apache.spark.sql.execution.datasources.parquet.ParquetFilters$$anonfun$createFilter$15.apply(ParquetFilters.scala:276)
at org.apache.spark.sql.execution.datasources.parquet.ParquetFilters$$anonfun$createFilter$15.apply(ParquetFilters.scala:276)
...
Contributor
There was a problem hiding this comment.
what about making this threshold configurable?
Member
There was a problem hiding this comment.
make it configurable. Use spark.sql.parquet.pushdown.inFilterThreshold. By default, it should be around 10. Please also check the perf.
cc @jiangxb1987
MemberAuthor
There was a problem hiding this comment.
It seems that the push-down performance is better when threshold is less than 300:
The code:
withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key ->"true") {
importtestImplicits._
withTempPath { path =>valtotal=10000000
(0 to total).toDF().coalesce(1)
.write.option("parquet.block.size", 512)
.parquet(path.getAbsolutePath)
valdf= spark.read.parquet(path.getAbsolutePath)
// scalastyle:off printlnvarlastSize=-1vari=16000while (i < total) {
valfilter=Range(0, total).filter(_ % i ==0)
i +=100if (lastSize != filter.size) {
if (lastSize ==-1) println(s"start size: ${filter.size}")
lastSize = filter.size
sql("set spark.sql.parquet.pushdown.inFilterThreshold=1000000")
valbegin1=System.currentTimeMillis()
df.where(s"id in(${filter.mkString(",")})").count()
valend1=System.currentTimeMillis()
valtime1= end1 - begin1
sql("set spark.sql.parquet.pushdown.inFilterThreshold=10")
valbegin2=System.currentTimeMillis()
df.where(s"id in(${filter.mkString(",")})").count()
valend2=System.currentTimeMillis()
valtime2= end2 - begin2
if (time1 <= time2) println(s"Max threshold: $lastSize")
}
}
}
}Contributor
There was a problem hiding this comment.
Thanks for doing this benchmark, this shall be useful, while I still have some questions:
- For small table (with columns <= 1000000), is the performance of InFilters still better than InSet?
- Can you also forge different filters? Currently your filters are distributed evenly, which don't always happen on real workload. We shall at least benchmark with different filter ratio (#rows filtered / #total rows)
MemberAuthor
There was a problem hiding this comment.
It mainly depends on how many row groups can skip. for small table (assuming only one row group). There is no obvious difference.
MemberAuthor
There was a problem hiding this comment.
I have prepared a test case that you can verify it:
test("Benchmark") {
defbenchmark(func: () =>Unit):Long= {
valstart=System.currentTimeMillis()
func()
valend=System.currentTimeMillis()
end - start
}
// scalastyle:off
withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_ENABLED.key ->"true") {
withTempPath { path =>Seq(1000000, 10000000).foreach { count =>Seq(1048576, 10485760, 104857600).foreach { blockSize =>
spark.range(count).toDF().selectExpr("id", "cast(id as string) as d1",
"cast(id as double) as d2", "cast(id as float) as d3", "cast(id as int) as d4",
"cast(id as decimal(38)) as d5")
.coalesce(1).write.mode("overwrite")
.option("parquet.block.size", blockSize).parquet(path.getAbsolutePath)
valdf= spark.read.parquet(path.getAbsolutePath)
println(s"path: ${path.getAbsolutePath}")
Seq(1000, 100, 10, 1).foreach { ratio =>
println(s"##########[ count: $count, blockSize: $blockSize, ratio: $ratio ]#########")
vari=1while (i <300) {
valfilter=Range(0, i).map(r => scala.util.Random.nextInt(count / ratio))
i +=4
sql("set spark.sql.parquet.pushdown.inFilterThreshold=1")
valvanillaTime= benchmark(() => df.where(s"id in(${filter.mkString(",")})").count())
sql("set spark.sql.parquet.pushdown.inFilterThreshold=1000")
valpushDownTime= benchmark(() => df.where(s"id in(${filter.mkString(",")})").count())
if (pushDownTime > vanillaTime) {
println(s"vanilla is better, threshold: ${filter.size}, $pushDownTime, $vanillaTime")
} else {
println(s"push down is better, threshold: ${filter.size}, $pushDownTime, $vanillaTime")
}
}
}
}
}
}
}
}SparkQA
commented
Jun 21, 2018
Test build #92156 has finished for PR 21603 at commit
|
wangyum
commented
Jun 21, 2018
MemberAuthor
Jenkins, retest this please. |
| case sources.In(name, values) if canMakeFilterOn(name) && values.length < 20 => | ||
| values.flatMap { v => | ||
| makeEq.lift(nameToType(name)).map(_(name, v)) | ||
| }.reduceLeftOption(FilterApi.or) |
Contributor
There was a problem hiding this comment.
what about null handling? Do we get the same result as before? Anyway, can we add a test for it?
SparkQA
commented
Jun 21, 2018
Test build #92161 has finished for PR 21603 at commit
|
HyukjinKwon
commented
Jun 21, 2018
Member
cc @gatorsmile, @viirya, @jiangxb1987 and @a10y too |
SparkQA
commented
Jun 22, 2018
Test build #92198 has finished for PR 21603 at commit
|
viirya
commented
Jun 22, 2018
Member
retest this please. |
| import testImplicits._ | ||
| withTempPath { path => | ||
| (0 to 1024).toDF("a").coalesce(1) |
| "This configuration only has an effect when 'spark.sql.parquet.filterPushdown' is enabled.") | ||
| .internal() | ||
| .intConf | ||
| .checkValue(threshold => threshold > 0 && threshold <= 300, |
Contributor
There was a problem hiding this comment.
why are we enforcing that the threshold <= 300 despite this might be a reasonable value, I think that the user should be able to set the value he/she wants. If the value the user sets is a bad one, its his/her problem to set a proper value. We can probably mention that from the experiments 300 seems to be a limit threshold in the description eventually, in order to give an hint to the user.
Member
There was a problem hiding this comment.
I also think to set a maximum value 300 is a bit weird.
| * Some utility function to convert Spark data source filters to Parquet filters. | ||
| */ | ||
| private[parquet] class ParquetFilters(pushDownDate: Boolean) { | ||
| private[parquet] class ParquetFilters(pushDownDate: Boolean, inFilterThreshold: Int) { |
Contributor
There was a problem hiding this comment.
nit: not a big deal, but since now SQLConf are available also on executor side, can we get its value inside the class rather than outside? If we add more configurations this constructor might explode...
viirya
reviewed
Jun 22, 2018
| val PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD = | ||
| buildConf("spark.sql.parquet.pushdown.inFilterThreshold") | ||
| .doc("The maximum number of values to filter push-down optimization for IN predicate. " + | ||
| "Large threshold will not provide much better performance. " + |
Member
There was a problem hiding this comment.
Large threshold won't necessarily provide much better performance.
SparkQA
commented
Jun 22, 2018
Test build #92203 has finished for PR 21603 at commit
|
SparkQA
commented
Jun 22, 2018
Test build #92210 has finished for PR 21603 at commit
|
| buildConf("spark.sql.parquet.pushdown.inFilterThreshold") | ||
| .doc("The maximum number of values to filter push-down optimization for IN predicate. " + | ||
| "Large threshold won't necessarily provide much better performance. " + | ||
| "The experiment argued that 300 is the limit threshold. " + |
Member
There was a problem hiding this comment.
This also depends on the data types, right?
MemberAuthor
There was a problem hiding this comment.
You are right.
| Type | limit threshold |
|---|---|
| string | 370 |
| int | 210 |
| long | 285 |
| double | 270 |
| float | 220 |
| decimal | Will not provide better performance |
Member
There was a problem hiding this comment.
An interesting finding. Thanks for the update. Maybe you do not need to mention this limit threshold in the doc?
How about post your finding in the PR description?
SparkQA
commented
Jun 25, 2018
Test build #92280 has finished for PR 21603 at commit
|
wangyum
commented
Jun 26, 2018
MemberAuthor
Jenkins, retest this please. |
SparkQA
commented
Jun 26, 2018
Test build #92340 has finished for PR 21603 at commit
|
wangyum
commented
Jun 30, 2018
MemberAuthor
Benchmark result: |
# Conflicts: # sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala # sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala # sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala # sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
SparkQA
commented
Jul 4, 2018
Test build #92605 has finished for PR 21603 at commit
|
HyukjinKwon
commented
Jul 5, 2018
Member
@wangyum, mind resolving the conflicts please? I think it's quite close. |
# Conflicts: # sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
viirya
reviewed
Jul 6, 2018
| Native ORC Vectorized 6840 / 8073 2.3434.9 1.1X | ||
| Native ORC Vectorized (Pushdown) 1041 / 107515.166.2 7.4X | ||
| Parquet Vectorized 7967 / 8175 2.0506.5 1.0X | ||
| Parquet Vectorized (Pushdown) 8155 / 84341.9518.5 1.0X |
Member
There was a problem hiding this comment.
I think this benchmark numbers are more meaningful if we show IN predicate is pushdown or not (or threshold).
SparkQA
commented
Jul 6, 2018
Test build #92666 has finished for PR 21603 at commit
|
wangyum
commented
Jul 6, 2018
MemberAuthor
retest this please. |
SparkQA
commented
Jul 6, 2018
Test build #92676 has finished for PR 21603 at commit
|
wangyum
commented
Jul 6, 2018
MemberAuthor
retest this please. |
SparkQA
commented
Jul 6, 2018
Test build #92684 has finished for PR 21603 at commit
|
wangyum
commented
Jul 7, 2018
MemberAuthor
retest this please. |
SparkQA
commented
Jul 7, 2018
Test build #92702 has finished for PR 21603 at commit
|
| "This configuration only has an effect when 'spark.sql.parquet.filterPushdown' is enabled.") | ||
| .internal() | ||
| .intConf | ||
| .checkValue(threshold => threshold > 0, "The threshold must be greater than 0.") |
Member
There was a problem hiding this comment.
Shell we allow, for example, -1 here to disable this?
MemberAuthor
There was a problem hiding this comment.
case sources.In(name, values) if canMakeFilterOn(name) && shouldConvertInPredicate(name) && values.distinct.length <= pushDownInFilterThreshold =>How about 0. values.distinct.length will not be less than 0.
| parquetFilters.createFilter(parquetSchema, sources.In("a", Array(10, 10))) | ||
| } | ||
| assertResult(Some(or( |
Member
There was a problem hiding this comment.
I think you can remove this test because it looks basically a duplicate of the below.
| .coalesce(1).write.option("parquet.block.size", 512) | ||
| .parquet(path.getAbsolutePath) | ||
| val df = spark.read.parquet(path.getAbsolutePath) | ||
| Seq(true, false).foreach { pushEnabled => |
Member
There was a problem hiding this comment.
@wangyum, does this really test the Parquet itself push down fine? I think you should stripSparkFilter and check if they are actually filtered out when spark.sql.parquet.filterPushdown is enabled.
MemberAuthor
There was a problem hiding this comment.
Updated to:
valactual= stripSparkFilter(df.where(filter)).collect().length
if (pushEnabled && count <= conf.parquetFilterPushDownInFilterThreshold) {
assert(actual >1&& actual < data.length)
} else {
assert(actual === data.length)
}HyukjinKwon
commented
Jul 13, 2018
Member
LGTM otherwise. |
wangyum
commented
Jul 13, 2018
| def canMakeFilterOn(name: String): Boolean = nameToType.contains(name) && !name.contains(".") | ||
| // All DataTypes that support `makeEq` can provide better performance. | ||
| def shouldConvertInPredicate(name: String): Boolean = nameToType(name) match { |
MemberAuthor
There was a problem hiding this comment.
@HyukjinKwon How about remove this?Timestamp type and Decimal type will be support soon.
Member
There was a problem hiding this comment.
It depends on which PR will be merged first. The corresponding PRs should update this.
Member
There was a problem hiding this comment.
Also need to update the benchmark suite.
| // converted (`ParquetFilters.createFilter` returns an `Option`). That's why a `flatMap` | ||
| // is used here. | ||
| .flatMap(new ParquetFilters(pushDownDate, pushDownStringStartWith) | ||
| .flatMap(new ParquetFilters(pushDownDate, pushDownStringStartWith, |
Member
There was a problem hiding this comment.
let us create val parquetFilters = new ParquetFilters(pushDownDate, pushDownStringStartWith, pushDownInFilterThreshold )
SparkQA
commented
Jul 14, 2018
Test build #92995 has finished for PR 21603 at commit
|
| assert(df.where(filter).count() === count) | ||
| val actual = stripSparkFilter(df.where(filter)).collect().length | ||
| if (pushEnabled && count <= conf.parquetFilterPushDownInFilterThreshold) { | ||
| assert(actual > 1 && actual < data.length) |
Member
There was a problem hiding this comment.
ah okay this tests block level filtering. lgtm
HyukjinKwon
commented
Jul 14, 2018
Member
Merged to master |
tanishgupta1
commented
May 29, 2019
I see this benchmarking is done for parquet filters. Did we have similar kind of benchmarking for ORC format too ? |
Member
Ask questions to mailing list. ORC is completely orthogonal with this Parquet PR. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for freeto join this conversation on GitHub.
Already have an account?
Sign in to comment
8 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
The original pr is: #18424
Add a new optimizer rule to convert an IN predicate to an equivalent Parquet filter and add
spark.sql.parquet.pushdown.inFilterThresholdto control limit thresholds. Different data types have different limit thresholds, this is a copy of data for reference:How was this patch tested?
unit tests and manual tests