Skip to content

[SPARK-39002][SQL] StringEndsWith/Contains support push down to Parquet #36328

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
WangGuangxin wants to merge 7 commits into from
Closed

[SPARK-39002][SQL] StringEndsWith/Contains support push down to Parquet #36328

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
e1a9def
push down StringEndsWith/Contains to parquet
WangGuangxin Apr 23, 2022
e97cd22
Fix ut & resolve comments
WangGuangxin Apr 24, 2022
45e13b6
update
WangGuangxin Apr 24, 2022
4ece85e
fix ut
WangGuangxin Apr 25, 2022
4c4e2c8
update ut
WangGuangxin Apr 26, 2022
e1a263d
add more test
WangGuangxin Apr 26, 2022
2a34968
fix build error
WangGuangxin Apr 27, 2022
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
13 changes: 11 additions & 2 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -959,6 +959,15 @@ object SQLConf {
.booleanConf
.createWithDefault(true)

val PARQUET_FILTER_PUSHDOWN_STRING_PREDICATE_ENABLED =
buildConf("spark.sql.parquet.filterPushdown.stringPredicate")

@jaceklaskowski jaceklaskowski Apr 23, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Since spark.sql.parquet.filterPushdown.string.startsWith is internal why not replacing it?

@WangGuangxin WangGuangxin Apr 24, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I'm afraid exising users who have already use it.

.doc("If true, enables Parquet filter push-down optimization for string predicate such " +
"as startsWith/endsWith/contains function. This configuration only has an effect when " +
s"'${PARQUET_FILTER_PUSHDOWN_ENABLED.key}' is enabled.")
.version("3.4.0")
.internal()
.fallbackConf(PARQUET_FILTER_PUSHDOWN_STRING_STARTSWITH_ENABLED)

val PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD =
buildConf("spark.sql.parquet.pushdown.inFilterThreshold")
.doc("For IN predicate, Parquet filter will push-down a set of OR clauses if its " +
Expand Down Expand Up @@ -4050,8 +4059,8 @@ class SQLConf extends Serializable with Logging {

def parquetFilterPushDownDecimal: Boolean = getConf(PARQUET_FILTER_PUSHDOWN_DECIMAL_ENABLED)

def parquetFilterPushDownStringStartWith: Boolean =
getConf(PARQUET_FILTER_PUSHDOWN_STRING_STARTSWITH_ENABLED)
def parquetFilterPushDownStringPredicate: Boolean =
getConf(PARQUET_FILTER_PUSHDOWN_STRING_PREDICATE_ENABLED)

def parquetFilterPushDownInFilterThreshold: Int =
getConf(PARQUET_FILTER_PUSHDOWN_INFILTERTHRESHOLD)
Expand Down
4 changes: 2 additions & 2 deletions ...src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -251,7 +251,7 @@ class ParquetFileFormat
val pushDownDate = sqlConf.parquetFilterPushDownDate
val pushDownTimestamp = sqlConf.parquetFilterPushDownTimestamp
val pushDownDecimal = sqlConf.parquetFilterPushDownDecimal
val pushDownStringStartWith = sqlConf.parquetFilterPushDownStringStartWith
val pushDownStringPredicate = sqlConf.parquetFilterPushDownStringPredicate
val pushDownInFilterThreshold = sqlConf.parquetFilterPushDownInFilterThreshold
val isCaseSensitive = sqlConf.caseSensitiveAnalysis
val parquetOptions = new ParquetOptions(options, sparkSession.sessionState.conf)
Expand Down Expand Up @@ -279,7 +279,7 @@ class ParquetFileFormat
pushDownDate,
pushDownTimestamp,
pushDownDecimal,
pushDownStringStartWith,
pushDownStringPredicate,
pushDownInFilterThreshold,
isCaseSensitive,
datetimeRebaseSpec)
Expand Down
34 changes: 32 additions & 2 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 @@ -48,7 +48,7 @@ class ParquetFilters(
pushDownDate: Boolean,
pushDownTimestamp: Boolean,
pushDownDecimal: Boolean,
pushDownStartWith: Boolean,
pushDownStringPredicate: Boolean,
pushDownInFilterThreshold: Int,
caseSensitive: Boolean,
datetimeRebaseSpec: RebaseSpec) {
Expand Down Expand Up @@ -747,7 +747,7 @@ class ParquetFilters(
}

case sources.StringStartsWith(name, prefix)
if pushDownStartWith && canMakeFilterOn(name, prefix) =>
if pushDownStringPredicate && canMakeFilterOn(name, prefix) =>
Option(prefix).map { v =>
FilterApi.userDefined(binaryColumn(nameToParquetField(name).fieldNames),
new UserDefinedPredicate[Binary] with Serializable {
Expand Down Expand Up @@ -778,6 +778,36 @@ class ParquetFilters(
)
}

case sources.StringEndsWith(name, suffix)
if pushDownStringPredicate && canMakeFilterOn(name, suffix) =>
Option(suffix).map { v =>
FilterApi.userDefined(binaryColumn(nameToParquetField(name).fieldNames),
new UserDefinedPredicate[Binary] with Serializable {
private val suffixStr = UTF8String.fromString(v)
override def canDrop(statistics: Statistics[Binary]): Boolean = false
override def inverseCanDrop(statistics: Statistics[Binary]): Boolean = false
override def keep(value: Binary): Boolean = {
value != null && UTF8String.fromBytes(value.getBytes).endsWith(suffixStr)
}
}
)
}

case sources.StringContains(name, value)
if pushDownStringPredicate && canMakeFilterOn(name, value) =>
Option(value).map { v =>
FilterApi.userDefined(binaryColumn(nameToParquetField(name).fieldNames),
new UserDefinedPredicate[Binary] with Serializable {
private val subStr = UTF8String.fromString(v)
override def canDrop(statistics: Statistics[Binary]): Boolean = false
override def inverseCanDrop(statistics: Statistics[Binary]): Boolean = false
override def keep(value: Binary): Boolean = {
value != null && UTF8String.fromBytes(value.getBytes).contains(subStr)
}
}
)
}

case _ => None
}
}
Expand Down
4 changes: 2 additions & 2 deletions ...org/apache/spark/sql/execution/datasources/v2/parquet/ParquetPartitionReaderFactory.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ case class ParquetPartitionReaderFactory(
private val pushDownDate = sqlConf.parquetFilterPushDownDate
private val pushDownTimestamp = sqlConf.parquetFilterPushDownTimestamp
private val pushDownDecimal = sqlConf.parquetFilterPushDownDecimal
private val pushDownStringStartWith = sqlConf.parquetFilterPushDownStringStartWith
private val pushDownStringPredicate = sqlConf.parquetFilterPushDownStringPredicate
private val pushDownInFilterThreshold = sqlConf.parquetFilterPushDownInFilterThreshold
private val datetimeRebaseModeInRead = options.datetimeRebaseModeInRead
private val int96RebaseModeInRead = options.int96RebaseModeInRead
Expand Down Expand Up @@ -221,7 +221,7 @@ case class ParquetPartitionReaderFactory(
pushDownDate,
pushDownTimestamp,
pushDownDecimal,
pushDownStringStartWith,
pushDownStringPredicate,
pushDownInFilterThreshold,
isCaseSensitive,
datetimeRebaseSpec)
Expand Down
4 changes: 2 additions & 2 deletions ...main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScanBuilder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ case class ParquetScanBuilder(
val pushDownDate = sqlConf.parquetFilterPushDownDate
val pushDownTimestamp = sqlConf.parquetFilterPushDownTimestamp
val pushDownDecimal = sqlConf.parquetFilterPushDownDecimal
val pushDownStringStartWith = sqlConf.parquetFilterPushDownStringStartWith
val pushDownStringPredicate = sqlConf.parquetFilterPushDownStringPredicate
val pushDownInFilterThreshold = sqlConf.parquetFilterPushDownInFilterThreshold
val isCaseSensitive = sqlConf.caseSensitiveAnalysis
val parquetSchema =
Expand All @@ -62,7 +62,7 @@ case class ParquetScanBuilder(
pushDownDate,
pushDownTimestamp,
pushDownDecimal,
pushDownStringStartWith,
pushDownStringPredicate,
pushDownInFilterThreshold,
isCaseSensitive,
// The rebase mode doesn't matter here because the filters are used to determine
Expand Down
3 changes: 2 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3072,7 +3072,8 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}

Seq("orc", "parquet").foreach { format =>
withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") {
withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "",
SQLConf.PARQUET_FILTER_PUSHDOWN_STRING_PREDICATE_ENABLED.key -> "false") {
withTempPath { dir =>
spark.range(10).map(i => (i, i.toString)).toDF("id", "s")
.write
Expand Down
32 changes: 32 additions & 0 deletions ...ore/src/test/scala/org/apache/spark/sql/execution/benchmark/FilterPushdownBenchmark.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,6 +242,38 @@ object FilterPushdownBenchmark extends SqlBasedBenchmark {
}
}

runBenchmark("Pushdown benchmark for StringEndsWith") {
withTempPath { dir =>
withTempTable("orcTable", "parquetTable") {
prepareStringDictTable(dir, numRows, 200, width)
Seq(
"value like '%10'",
"value like '%1000'",
s"value like '%${mid.toString.substring(0, mid.toString.length - 1)}'"
).foreach { whereExpr =>
val title = s"StringEndsWith filter: ($whereExpr)"
filterPushDownBenchmark(numRows, title, whereExpr)
}
}
}
}

runBenchmark("Pushdown benchmark for StringContains") {
withTempPath { dir =>
withTempTable("orcTable", "parquetTable") {
prepareStringDictTable(dir, numRows, 200, width)
Seq(
"value like '%10%'",
"value like '%1000%'",
s"value like '%${mid.toString.substring(0, mid.toString.length - 1)}%'"
).foreach { whereExpr =>
val title = s"StringContains filter: ($whereExpr)"
filterPushDownBenchmark(numRows, title, whereExpr)
}
}
}
}

runBenchmark(s"Pushdown benchmark for ${DecimalType.simpleString}") {
withTempPath { dir =>
Seq(
Expand Down
83 changes: 70 additions & 13 deletions ...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 @@ -81,7 +81,7 @@ abstract class ParquetFilterSuite extends QueryTest with ParquetTest with Shared
datetimeRebaseSpec: RebaseSpec = RebaseSpec(LegacyBehaviorPolicy.CORRECTED)
): ParquetFilters =
new ParquetFilters(schema, conf.parquetFilterPushDownDate, conf.parquetFilterPushDownTimestamp,
conf.parquetFilterPushDownDecimal, conf.parquetFilterPushDownStringStartWith,
conf.parquetFilterPushDownDecimal, conf.parquetFilterPushDownStringPredicate,
conf.parquetFilterPushDownInFilterThreshold,
caseSensitive.getOrElse(conf.caseSensitiveAnalysis),
datetimeRebaseSpec)
Expand Down Expand Up @@ -207,20 +207,24 @@ abstract class ParquetFilterSuite extends QueryTest with ParquetTest with Shared
}
}

// This function tests that exactly go through the `canDrop` and `inverseCanDrop`.
private def testStringStartsWith(dataFrame: DataFrame, filter: String): Unit = {
// This function tests that exactly go through the `keep`, `canDrop` and `inverseCanDrop`.
private def testStringPredicate(dataFrame: DataFrame, filter: String,
shouldFilterOut: Boolean, enableDictionary: Boolean = true): Unit = {
withTempPath { dir =>
val path = dir.getCanonicalPath
dataFrame.write.option("parquet.block.size", 512).parquet(path)
dataFrame.write
.option("parquet.block.size", 512)
.option(ParquetOutputFormat.ENABLE_DICTIONARY, enableDictionary)
.parquet(path)
Seq(true, false).foreach { pushDown =>
withSQLConf(
SQLConf.PARQUET_FILTER_PUSHDOWN_STRING_STARTSWITH_ENABLED.key -> pushDown.toString) {
SQLConf.PARQUET_FILTER_PUSHDOWN_STRING_PREDICATE_ENABLED.key -> pushDown.toString) {
val accu = new NumRowGroupsAcc
sparkContext.register(accu)

val df = spark.read.parquet(path).filter(filter)
df.foreachPartition((it: Iterator[Row]) => it.foreach(v => accu.add(0)))
if (pushDown) {
if (pushDown && shouldFilterOut) {
assert(accu.value == 0)
} else {
assert(accu.value > 0)
Expand Down Expand Up @@ -970,7 +974,12 @@ abstract class ParquetFilterSuite extends QueryTest with ParquetTest with Shared
))

val parquetSchema = new SparkToParquetSchemaConverter(conf).convert(schema)
val parquetFilters = createParquetFilters(parquetSchema)
// Following tests are used to check one arm of AND/OR can't be pushed down,
// so we disable string predicate pushdown here
var parquetFilters: ParquetFilters = null
withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_STRING_PREDICATE_ENABLED.key -> "false") {
parquetFilters = createParquetFilters(parquetSchema)
}
assertResult(Some(and(
lt(intColumn("a"), 10: Integer),
gt(doubleColumn("c"), 1.5: java.lang.Double)))
Expand Down Expand Up @@ -1114,7 +1123,12 @@ abstract class ParquetFilterSuite extends QueryTest with ParquetTest with Shared
))

val parquetSchema = new SparkToParquetSchemaConverter(conf).convert(schema)
val parquetFilters = createParquetFilters(parquetSchema)
// Following tests are used to check one arm of AND/OR can't be pushed down,
// so we disable string predicate pushdown here
var parquetFilters: ParquetFilters = null
withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_STRING_PREDICATE_ENABLED.key -> "false") {
parquetFilters = createParquetFilters(parquetSchema)
}
// Testing
// case sources.Or(lhs, rhs) =>
// ...
Expand Down Expand Up @@ -1169,7 +1183,12 @@ abstract class ParquetFilterSuite extends QueryTest with ParquetTest with Shared
))

val parquetSchema = new SparkToParquetSchemaConverter(conf).convert(schema)
val parquetFilters = createParquetFilters(parquetSchema)
// Following tests are used to check one arm of AND/OR can't be pushed down,
// so we disable string predicate pushdown here
var parquetFilters: ParquetFilters = null
withSQLConf(SQLConf.PARQUET_FILTER_PUSHDOWN_STRING_PREDICATE_ENABLED.key -> "false") {
parquetFilters = createParquetFilters(parquetSchema)
}
assertResult(Seq(sources.And(sources.LessThan("a", 10), sources.GreaterThan("c", 1.5D)))) {
parquetFilters.convertibleFilters(
Seq(sources.And(
Expand Down Expand Up @@ -1476,12 +1495,50 @@ abstract class ParquetFilterSuite extends QueryTest with ParquetTest with Shared
classOf[UserDefinedByInstance[_, _]],
Seq.empty[Row])
}
}

test("filter pushdown - StringPredicate") {
import testImplicits._
// Test canDrop() has taken effect
testStringStartsWith(spark.range(1024).map(_.toString).toDF(), "value like 'a%'")
// Test inverseCanDrop() has taken effect
testStringStartsWith(spark.range(1024).map(c => "100").toDF(), "value not like '10%'")
// keep() should take effect on StartsWith/EndsWith/Contains
Seq(
"value like 'a%'", // StartsWith
"value like '%a'", // EndsWith
"value like '%a%'" // Contains

@sadikovi sadikovi May 4, 2022 •
edited
Loading

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

A quick comment. How does this verify the "keep()" test? Shouldn't it also be "canDrop()"?

@sadikovi sadikovi May 4, 2022

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Does this test assume that dictionary filtering is enabled or not?

@sadikovi sadikovi May 4, 2022 •
edited
Loading

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

I think the test is buggy and does not reflect the actual implementation of the filter. NumRowGroupAcc does not actually count row groups, it counts the number of records passed through the filter. For example, for the contains filter we should still read all of the row groups.

Example of the log:

[canDrop] statistics=org.apache.parquet.filter2.predicate.Statistics@52cd90cd => false
[canDrop] statistics=org.apache.parquet.filter2.predicate.Statistics@64e59f7e => false
  [keep] statistics=Binary{1 constant bytes, [49]} => false
  [keep] statistics=Binary{1 constant bytes, [50]} => false
  [keep] statistics=Binary{1 constant bytes, [51]} => false
  [keep] statistics=Binary{1 constant bytes, [52]} => false
...

Can the author update the test to reflect the implementation? cc @cloud-fan @sunchao.
You may need to enforce things like row group/dictionary filtering as well as record level filtering.

@WangGuangxin WangGuangxin May 4, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Hi @sadikovi , the NumRowGroupsAcc is the actually filtered row groups, you can find it here

spark/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/SpecificParquetRecordReaderBase.java

Line 130 in a1aa200

if (accu.isDefined() && accu.get().getClass().getSimpleName().equals("NumRowGroupsAcc")) {
.

As to the keep() test, the dictionary filter is enabled and there are duplicated records in test data, so parquet will generate dictionary when writing data and dictionary filter is used when reading it.

When we test canDrop, the test data has no duplicate so there is no dictionary generated in parquet, statistics row group filter is used which will call canDrop.

Correct me if I'm wrong.

@sadikovi sadikovi May 4, 2022

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Yes, that was my point - the test needs to be updated to make sure dictionary pages are written and the dictionary filtering is enabled. Without it, the test does not verify the implementation.

@sadikovi sadikovi May 4, 2022 •
edited
Loading

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Can you open a follow-up PR to update the test? You can explicitly enable dictionary filtering in the test for the "keep" part of the test to highlight that the test passes due to dictionary filtering, otherwise it could be confusing for people.

@WangGuangxin WangGuangxin May 4, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Maybe the fourth param in testStringPredicate is what you need?

spark/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala

Line 217 in a1aa200

.option(ParquetOutputFormat.ENABLE_DICTIONARY, enableDictionary)

It's enabled(by default) in keep test

spark/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala

Line 1532 in a1aa200

true)

and disabled in canDrop test

spark/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala

Line 1546 in a1aa200

enableDictionary = false)

@sadikovi sadikovi May 5, 2022

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Enabling dictionary does not control dictionary filtering, there is a separate flag for it.

@cloud-fan cloud-fan May 5, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Given the discussions here, seems this is not a simple thing. @sadikovi can you open a followup PR directly to demonstrate your idea?

).foreach { filter =>
testStringPredicate(
// dictionary will be generated since there are duplicated values
spark.range(1000).map(t => (t % 10).toString).toDF(),
filter,
true)
}

// canDrop() should take effect on StartsWith,
// and has no effect on EndsWith/Contains
Seq(
("value like 'a%'", true), // StartsWith
("value like '%a'", false), // EndsWith
("value like '%a%'", false) // Contains
).foreach { case (filter, shouldFilterOut) =>
testStringPredicate(
spark.range(1024).map(_.toString).toDF(),
filter,
shouldFilterOut,
enableDictionary = false)
}

// inverseCanDrop() should take effect on StartsWith,
// and has no effect on EndsWith/Contains
Seq(
("value not like '10%'", true), // StartsWith
("value not like '%10'", false), // EndsWith
("value not like '%10%'", false) // Contains
).foreach { case (filter, shouldFilterOut) =>
testStringPredicate(
spark.range(1024).map(c => "100").toDF(),
filter,
shouldFilterOut,
enableDictionary = false)
}
}

test("SPARK-17091: Convert IN predicate to Parquet filter push-down") {
Expand Down