Skip to content

[SPARK-25559][SQL] Remove the unsupported predicates in Parquet when possible #22574

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dbtsai wants to merge 5 commits into from
Closed

[SPARK-25559][SQL] Remove the unsupported predicates in Parquet when possible #22574

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8c76e31
Remove unsupported predicates in parquet
dbtsai Sep 27, 2018
7792e99
Add a nested And test case
dongjoon-hyun Sep 28, 2018
29a9aa0
Merge pull request #5 from dongjoon-hyun/PR-22574
dbtsai Sep 28, 2018
d37b2e8
address feedback
dbtsai Sep 28, 2018
9a9e47f
Added more test and doc
dbtsai Sep 28, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 19 additions & 14 deletions ...re/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -394,7 +394,13 @@ private[parquet] class ParquetFilters(
*/
def createFilter(schema: MessageType, predicate: sources.Filter): Option[FilterPredicate] = {
val nameToParquetField = getFieldMap(schema)
createFilterHelper(nameToParquetField, predicate, canRemoveOneSideInAnd = true)
}

private def createFilterHelper(
nameToParquetField: Map[String, ParquetField],
predicate: sources.Filter,
canRemoveOneSideInAnd: Boolean): Option[FilterPredicate] = {
// Decimal type must make sure that filter value's scale matched the file.
// If doesn't matched, which would cause data corruption.
def isDecimalMatched(value: Any, decimalMeta: DecimalMetadata): Boolean = value match {
Expand Down Expand Up @@ -488,26 +494,25 @@ private[parquet] class ParquetFilters(
.map(_(nameToParquetField(name).fieldName, value))

case sources.And(lhs, rhs) =>
// At here, it is not safe to just convert one side if we do not understand the
// other side. Here is an example used to explain the reason.
// Let's say we have NOT(a = 2 AND b in ('1')) and we do not understand how to
// convert b in ('1'). If we only convert a = 2, we will end up with a filter
// NOT(a = 2), which will generate wrong results.
// Pushing one side of AND down is only safe to do at the top level.
// You can see ParquetRelation's initializeLocalJobFunc method as an example.
Copy link
Member

@viirya viirya Sep 28, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This example is good to show the cases we can't remove one side. Can we still keep it?

Copy link
Contributor

@cloud-fan cloud-fan Sep 28, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1

Copy link
Member

@dongjoon-hyun dongjoon-hyun Sep 28, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1

Copy link
Member Author

@dbtsai dbtsai Sep 28, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

addressed and added more tests.

for {
lhsFilter <- createFilter(schema, lhs)
rhsFilter <- createFilter(schema, rhs)
} yield FilterApi.and(lhsFilter, rhsFilter)
// If the unsupported predicate is in the top level `And` condition or in the child
// `And` condition before hitting `Not` or `Or` condition, it can be safely removed.
(createFilterHelper(nameToParquetField, lhs, canRemoveOneSideInAnd = true),
Copy link
Contributor

@cloud-fan cloud-fan Sep 28, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

a bit about style

val lhs = createFilterHelper...
val rhs = createFilterHelper...
(lhs, rhs) match {
  ...
}

Copy link
Member Author

@dbtsai dbtsai Sep 28, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed. Thanks.

createFilterHelper(nameToParquetField, rhs, canRemoveOneSideInAnd = true)) match {
Copy link
Member

@dongjoon-hyun dongjoon-hyun Sep 28, 2018 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you for pinging me, @dbtsai .

Here, instead of canRemoveOneSideInAnd = true, we need to use the received canRemoveOneSideInAnd, don't we? Otherwise, this will cause a bug in the nested And clauses. So, could you review and merge the test case, dbtsai#5, and fix the PR accordingly?

Copy link
Contributor

@cloud-fan cloud-fan Sep 28, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1

Copy link
Member Author

@dbtsai dbtsai Sep 28, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for catching this. I just fixed it.

case (Some(lhsFilter), Some(rhsFilter)) => Some(FilterApi.and(lhsFilter, rhsFilter))
case (Some(lhsFilter), None) if canRemoveOneSideInAnd => Some(lhsFilter)
case (None, Some(rhsFilter)) if canRemoveOneSideInAnd => Some(rhsFilter)
case _ => None
}

case sources.Or(lhs, rhs) =>
for {
lhsFilter <- createFilter(schema, lhs)
rhsFilter <- createFilter(schema, rhs)
lhsFilter <- createFilterHelper(nameToParquetField, lhs, canRemoveOneSideInAnd = false)
rhsFilter <- createFilterHelper(nameToParquetField, rhs, canRemoveOneSideInAnd = false)
} yield FilterApi.or(lhsFilter, rhsFilter)

case sources.Not(pred) =>
createFilter(schema, pred).map(FilterApi.not)
createFilterHelper(nameToParquetField, pred, canRemoveOneSideInAnd = false)
.map(FilterApi.not)

case sources.In(name, values) if canMakeFilterOn(name, values.head)
&& values.distinct.length <= pushDownInFilterThreshold =>
Expand Down
13 changes: 12 additions & 1 deletion ...rc/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -770,14 +770,25 @@ class ParquetFilterSuite extends QueryTest with ParquetTest with SharedSQLContex
sources.GreaterThan("c", 1.5D)))
}

assertResult(None) {
assertResult(Some(lt(intColumn("a"), 10: Integer))) {
parquetFilters.createFilter(
parquetSchema,
sources.And(
sources.LessThan("a", 10),
sources.StringContains("b", "prefix")))
}

assertResult(None) {
parquetFilters.createFilter(
parquetSchema,
sources.Not(
sources.And(
sources.And(
sources.GreaterThan("a", 1),
sources.StringContains("b", "prefix")),
sources.GreaterThan("a", 2))))
}

assertResult(None) {
parquetFilters.createFilter(
parquetSchema,
Expand Down