[SPARK-17091][SQL] Add rule to convert IN predicate to equivalent Parquet filter #21603
There are no files selected for viewing
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
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
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -37,7 +37,10 @@ import org.apache.spark.unsafe.types.UTF8String | |
| /** | ||
| * Some utility function to convert Spark data source filters to Parquet filters. | ||
| */ | ||
| private[parquet] class ParquetFilters( | ||
| pushDownDate: Boolean, | ||
| pushDownStartWith: Boolean, | ||
| pushDownInFilterThreshold: Int) { | ||
|
|
||
| private case class ParquetSchemaType( | ||
| originalType: OriginalType, | ||
|
|
@@ -232,6 +235,15 @@ private[parquet] class ParquetFilters(pushDownDate: Boolean, pushDownStartWith: | |
| // See SPARK-20364. | ||
| 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 { | ||
|
Member
Author
There was a problem hiding this comment. @HyukjinKwon How about remove this?
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. |
||
| case ParquetBooleanType | ParquetByteType | ParquetShortType | ParquetIntegerType | ||
| | ParquetLongType | ParquetFloatType | ParquetDoubleType | ParquetStringType | ||
| | ParquetBinaryType => true | ||
| case ParquetDateType if pushDownDate => true | ||
| case _ => false | ||
| } | ||
|
|
||
| // NOTE: | ||
| // | ||
| // For any comparison operator `cmp`, both `a cmp NULL` and `NULL cmp a` evaluate to `NULL`, | ||
|
|
@@ -295,6 +307,12 @@ private[parquet] class ParquetFilters(pushDownDate: Boolean, pushDownStartWith: | |
| case sources.Not(pred) => | ||
| createFilter(schema, pred).map(FilterApi.not) | ||
|
|
||
| case sources.In(name, values) if canMakeFilterOn(name) && shouldConvertInPredicate(name) | ||
| && values.distinct.length <= pushDownInFilterThreshold => | ||
| values.distinct.flatMap { v => | ||
| makeEq.lift(nameToType(name)).map(_(name, v)) | ||
| }.reduceLeftOption(FilterApi.or) | ||
|
|
||
| case sources.StringStartsWith(name, prefix) if pushDownStartWith && canMakeFilterOn(name) => | ||
| Option(prefix).map { v => | ||
| FilterApi.userDefined(binaryColumn(name), | ||
|
|
||
Oops, something went wrong.
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.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This also depends on the data types, right?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You are right.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
See the comment #21603 (comment)