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

WangGuangxin · 2022-04-23T08:03:05Z

What changes were proposed in this pull request?

Push down StringEndsWith/Contains to Parquet so that we can leverage Parquet Dictionary Filtering

Why are the changes needed?

Improve performance.

FilterPushDownBenchmark:

================================================================================================
Pushdown benchmark for StringEndsWith
================================================================================================

OpenJDK 64-Bit Server VM 1.8.0_292-b10 on Mac OS X 10.16
Intel(R) Core(TM) i7-1068NG7 CPU @ 2.30GHz
StringEndsWith filter: (value like '%10'):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
-------------------------------------------------------------------------------------------------------------------------
Parquet Vectorized                                  7666           7771         117          2.1         487.4       1.0X
Parquet Vectorized (Pushdown)                        540            554          18         29.1          34.3      14.2X
Native ORC Vectorized                               8206           8417         203          1.9         521.7       0.9X
Native ORC Vectorized (Pushdown)                    8120           8674         422          1.9         516.2       0.9X

OpenJDK 64-Bit Server VM 1.8.0_292-b10 on Mac OS X 10.16
Intel(R) Core(TM) i7-1068NG7 CPU @ 2.30GHz
StringEndsWith filter: (value like '%1000'):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
---------------------------------------------------------------------------------------------------------------------------
Parquet Vectorized                                    7007           7122         224          2.2         445.5       1.0X
Parquet Vectorized (Pushdown)                          423            485          92         37.2          26.9      16.6X
Native ORC Vectorized                                 7368           7629         373          2.1         468.5       1.0X
Native ORC Vectorized (Pushdown)                      7998           8349         270          2.0         508.5       0.9X

OpenJDK 64-Bit Server VM 1.8.0_292-b10 on Mac OS X 10.16
Intel(R) Core(TM) i7-1068NG7 CPU @ 2.30GHz
StringEndsWith filter: (value like '%786432'):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
-----------------------------------------------------------------------------------------------------------------------------
Parquet Vectorized                                      7012           7210         238          2.2         445.8       1.0X
Parquet Vectorized (Pushdown)                            419            431          14         37.6          26.6      16.7X
Native ORC Vectorized                                   7513           7995         447          2.1         477.6       0.9X
Native ORC Vectorized (Pushdown)                        8310           8811         448          1.9         528.3       0.8X


================================================================================================
Pushdown benchmark for StringContains
================================================================================================

OpenJDK 64-Bit Server VM 1.8.0_292-b10 on Mac OS X 10.16
Intel(R) Core(TM) i7-1068NG7 CPU @ 2.30GHz
StringContains filter: (value like '%10%'):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
--------------------------------------------------------------------------------------------------------------------------
Parquet Vectorized                                   7588           8125         328          2.1         482.4       1.0X
Parquet Vectorized (Pushdown)                        1029           1068          25         15.3          65.4       7.4X
Native ORC Vectorized                                7803           7859          92          2.0         496.1       1.0X
Native ORC Vectorized (Pushdown)                     8944           9443         459          1.8         568.6       0.8X

OpenJDK 64-Bit Server VM 1.8.0_292-b10 on Mac OS X 10.16
Intel(R) Core(TM) i7-1068NG7 CPU @ 2.30GHz
StringContains filter: (value like '%1000%'):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
----------------------------------------------------------------------------------------------------------------------------
Parquet Vectorized                                     7476           8343         710          2.1         475.3       1.0X
Parquet Vectorized (Pushdown)                           424            427           2         37.1          27.0      17.6X
Native ORC Vectorized                                  7503           8261         818          2.1         477.0       1.0X
Native ORC Vectorized (Pushdown)                       8124           8609         548          1.9         516.5       0.9X

OpenJDK 64-Bit Server VM 1.8.0_292-b10 on Mac OS X 10.16
Intel(R) Core(TM) i7-1068NG7 CPU @ 2.30GHz
StringContains filter: (value like '%786432%'):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------------
Parquet Vectorized                                       7070           7274         199          2.2         449.5       1.0X
Parquet Vectorized (Pushdown)                             441            478          32         35.6          28.1      16.0X
Native ORC Vectorized                                    7564           7937         323          2.1         480.9       0.9X
Native ORC Vectorized (Pushdown)                         8623           8921         228          1.8         548.2       0.8X

Does this PR introduce any user-facing change?

No

How was this patch tested?

added UT

WangGuangxin · 2022-04-23T08:07:29Z

cc @wangyum @cloud-fan

jaceklaskowski · 2022-04-23T11:46:29Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

    .createWithDefault(true)

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


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

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

AmplabJenkins · 2022-04-23T15:13:30Z

Can one of the admins verify this patch?

viirya · 2022-04-24T04:52:10Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala

        }

+      case sources.StringEndsWith(name, prefix)
+        if pushDownStringPredicate && canMakeFilterOn(name, prefix) =>


nit: indent, following L750.

viirya · 2022-04-24T04:52:20Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala

+        }
+
+      case sources.StringContains(name, value)
+        if pushDownStringPredicate && canMakeFilterOn(name, value) =>


viirya · 2022-04-24T04:55:03Z

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala

+        Option(prefix).map { v =>
+          FilterApi.userDefined(binaryColumn(nameToParquetField(name).fieldNames),
+            new UserDefinedPredicate[Binary] with Serializable {
+              private val strToBinary = Binary.fromReusedByteArray(v.getBytes)


You can just keep UTF8String.fromBytes(strToBinary.getBytes) instead doing it every time in keep.

good catch. updated

viirya

There are a few test failures in ParquetV2FilterSuite.

viirya · 2022-04-24T04:58:38Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

+      .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 " +
+        "'${PARQUET_FILTER_PUSHDOWN_ENABLED.key}' is enabled.")
+      .version("3.3.0")


viirya · 2022-04-24T16:50:14Z

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

+    }
+  }
+
+  test("filter pushdown - StringEndsWith/Contains") {


don't need to test StringStartsWith here?

StringStartsWith push down is an existing feature and has been tested at L1426

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

Line 1426 in 5046b8c

test("filter pushdown - StringStartsWith") {

Oh, I mean this test specially does test with testStringPredicateWithDictionaryFilter, but I don't see StringStartsWith is included here. Don't we need to do testStringPredicateWithDictionaryFilter for it?

can we merge the existing startswith test into the new one?

HyukjinKwon · 2022-04-25T08:55:56Z

cc @wangyum

viirya · 2022-04-26T02:39:16Z

lgtm if CI can pass.

viirya · 2022-04-26T02:39:49Z

cc @huaxingao

huaxingao · 2022-04-26T05:46:28Z

Looks good to me overall. Do we need a test similar to this one test("filter pushdown - StringStartsWith") to make sure the the filters for StringEndsWith and StringContains are created and pushed down to Scan OK?

WangGuangxin · 2022-04-26T09:44:46Z

Looks good to me overall. Do we need a test similar to this one test("filter pushdown - StringStartsWith") to make sure the the filters for StringEndsWith and StringContains are created and pushed down to Scan OK?

done

viirya · 2022-04-26T17:29:30Z

@WangGuangxin can you fix the error in "Scala 2.13 build with SBT"?

[error] /home/runner/work/spark/spark/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala:1447:57: procedure syntax is deprecated: instead, add `: Unit =` to explicitly declare `checkStringFilterPushdown`'s return type
[error]       sourceFilter: (String, String) => sources.Filter) {

WangGuangxin · 2022-04-27T03:25:53Z

@WangGuangxin can you fix the error in "Scala 2.13 build with SBT"?

[error] /home/runner/work/spark/spark/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala:1447:57: procedure syntax is deprecated: instead, add `: Unit =` to explicitly declare `checkStringFilterPushdown`'s return type
[error]       sourceFilter: (String, String) => sources.Filter) {

done

viirya · 2022-04-27T04:12:00Z

Thanks. Merging to master.

sadikovi · 2022-05-04T05:55:42Z

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

+    Seq(
+      "value like 'a%'", // StartsWith
+      "value like '%a'", // EndsWith
+      "value like '%a%'" // Contains


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

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

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.

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.

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.

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.

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)

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

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

### What changes were proposed in this pull request? This PR updates `FilterPushdownBenchmark` results. ### Why are the changes needed? These PRs do not or do not fully update the `FilterPushdownBenchmark` results: #36328 #36629 #36892 ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? N/A. Closes #37022 from wangyum/SPARK-39631. Authored-by: Yuming Wang <[email protected]> Signed-off-by: Yuming Wang <[email protected]>

push down StringEndsWith/Contains to parquet

e1a9def

github-actions bot added the SQL label Apr 23, 2022

jaceklaskowski reviewed Apr 23, 2022

View reviewed changes

viirya reviewed Apr 24, 2022

View reviewed changes

WangGuangxin added 2 commits April 24, 2022 23:22

Fix ut & resolve comments

e97cd22

update

45e13b6

viirya reviewed Apr 24, 2022

View reviewed changes

fix ut

4ece85e

HyukjinKwon changed the title ~~[SPARK-39002][SQL]StringEndsWith/Contains support push down to Parquet~~ [SPARK-39002][SQL] StringEndsWith/Contains support push down to Parquet Apr 25, 2022

update ut

4c4e2c8

viirya approved these changes Apr 26, 2022

View reviewed changes

cloud-fan approved these changes Apr 26, 2022

View reviewed changes

add more test

e1a263d

huaxingao approved these changes Apr 26, 2022

View reviewed changes

fix build error

2a34968

viirya closed this in 1b7c636 Apr 27, 2022

sadikovi reviewed May 4, 2022

View reviewed changes

wangyum mentioned this pull request Jun 29, 2022

[SPARK-39631][SQL][TESTS] Update FilterPushdownBenchmark results #37022

Closed

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

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

Uh oh!

Conversation

WangGuangxin commented Apr 23, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

WangGuangxin commented Apr 23, 2022

Uh oh!

jaceklaskowski Apr 23, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AmplabJenkins commented Apr 23, 2022

Uh oh!

viirya Apr 24, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HyukjinKwon commented Apr 25, 2022

Uh oh!

viirya commented Apr 26, 2022

Uh oh!

viirya commented Apr 26, 2022

Uh oh!

huaxingao commented Apr 26, 2022

Uh oh!

WangGuangxin commented Apr 26, 2022

Uh oh!

viirya commented Apr 26, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

WangGuangxin commented Apr 27, 2022

Uh oh!

viirya commented Apr 27, 2022

Uh oh!

sadikovi May 4, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sadikovi May 4, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

WangGuangxin commented Apr 23, 2022 •

edited

Loading

jaceklaskowski Apr 23, 2022 •

edited

Loading

viirya Apr 24, 2022 •

edited

Loading

viirya commented Apr 26, 2022 •

edited

Loading

sadikovi May 4, 2022 •

edited

Loading

sadikovi May 4, 2022 •

edited

Loading

sadikovi May 4, 2022 •

edited

Loading