[SPARK-27473][SQL] Support filter push down for status fields in binary file data source #24387

WeichenXu123 · 2019-04-17T01:21:32Z

What changes were proposed in this pull request?

Support 5 kinds of filters with and/or:

LessThan
LessThanOrEqual
GreatThan
GreatThanOrEqual
EqualTo

Support filters applied on 2 columns:

modificationTime
length

Note:
In order to support datasource filter push-down, I flatten schema to be:

val schema = StructType(
    StructField("path", StringType, false) ::
    StructField("modificationTime", TimestampType, false) ::
    StructField("length", LongType, false) ::
    StructField("content", BinaryType, true) :: Nil)

How was this patch tested?

To be added.

Please review http://spark.apache.org/contributing.html before opening a pull request.

SparkQA · 2019-04-17T01:27:09Z

Test build #104639 has finished for PR 24387 at commit 83380a7.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

.../src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala

mengxr · 2019-04-18T04:03:11Z

.../src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala

+
+  private[binaryfile] def createFilterFunctions(filter: Filter): Seq[FileStatus => Boolean] = {
+    filter match {
+      case andFilter: And =>


case And(left, right) =>

shall we support Or?

mengxr · 2019-04-18T04:03:36Z

.../src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala

+    filter match {
+      case andFilter: And =>
+        createFilterFunctions(andFilter.left) ++ createFilterFunctions(andFilter.right)
+      case LessThan("length", value: Long) =>


Instead of group by filter, could you group them by the field name?

mengxr · 2019-04-18T04:03:47Z

.../src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala

+      case andFilter: And =>
+        createFilterFunctions(andFilter.left) ++ createFilterFunctions(andFilter.right)
+      case LessThan("length", value: Long) =>
+        Seq((status: FileStatus) => status.getLen < value)


minor: Seq => Some

mengxr · 2019-04-18T04:05:29Z

.../src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala

+        Seq((status: FileStatus) => status.getLen < value)
+      case LessThan("modificationTime", value: Timestamp) =>
+        Seq((status: FileStatus) => status.getModificationTime < value.getTime)
+      case LessThanOrEqual("length", value: Long) =>


Just to confirm, does SQL always convert the value to Long?

yes, if the data type is long type.

mengxr · 2019-04-18T04:05:39Z

.../src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala

+        Seq((status: FileStatus) => status.getLen >= value)
+      case GreaterThanOrEqual("modificationTime", value: Timestamp) =>
+        Seq((status: FileStatus) => status.getModificationTime >= value.getTime)
+      case _ => Seq.empty


minor: => None

mengxr · 2019-04-18T04:12:13Z

...test/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormatSuite.scala

        col("year") // this is a partition column
      )
+    if (sqlFilter != null) {
+      resultDF = resultDF.filter(sqlFilter)


The test doesn't really test filter push down because without it the test still passes. I think we should unit test the buildReader function alone.

mengxr · 2019-04-18T04:13:28Z

...test/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormatSuite.scala

  }

+
+  def testCreateFilterFunctions(


See comment above. We can just unit test the buildReader func, which would look similar to your test here.

...test/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormatSuite.scala

mengxr · 2019-04-18T04:18:16Z

@WeichenXu123 Can you also handle EqualTo? Are we expecting Or and Not? cc: @cloud-fan

SparkQA · 2019-04-18T06:27:43Z

Test build #104688 has finished for PR 24387 at commit 8acd36e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-04-19T07:05:01Z

Test build #104743 has finished for PR 24387 at commit ecc156d.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

WeichenXu123 · 2019-04-19T17:51:17Z

Jenkins, retest this please.

SparkQA · 2019-04-19T21:07:22Z

Test build #104757 has finished for PR 24387 at commit ecc156d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-04-20T07:05:02Z

Test build #104772 has finished for PR 24387 at commit a4aba74.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

mengxr · 2019-04-20T08:13:07Z

WeichenXu123#7

Change to unit test `createFilterFunction`, which should be sufficient.

WeichenXu123 · 2019-04-20T18:09:23Z

Thanks @mengxr 's update! Looks good to me.

mengxr · 2019-04-20T18:58:47Z

@WeichenXu123 I pushed some changes to your branch that added more tests.

mengxr · 2019-04-20T18:59:35Z

LGTM pending Jenkins. @cloud-fan Do you want to make a pass?

SparkQA · 2019-04-20T21:01:49Z

Test build #104777 has finished for PR 24387 at commit 600aa9a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-04-20T21:17:29Z

Test build #104778 has finished for PR 24387 at commit 7d9bdb0.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-04-20T21:51:38Z

Test build #104779 has finished for PR 24387 at commit 71db855.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HyukjinKwon · 2019-04-28T15:47:54Z

.../src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala

+      case GreaterThanOrEqual(MODIFICATION_TIME, value: Timestamp) =>
+        _.getModificationTime >= value.getTime
+      case EqualTo(MODIFICATION_TIME, value: Timestamp) =>
+        _.getModificationTime == value.getTime


Looks EqualNullSafe and In are missing.

…ry file data source ## What changes were proposed in this pull request? Support 4 kinds of filters: - LessThan - LessThanOrEqual - GreatThan - GreatThanOrEqual Support filters applied on 2 columns: - modificationTime - length Note: In order to support datasource filter push-down, I flatten schema to be: ``` val schema = StructType( StructField("path", StringType, false) :: StructField("modificationTime", TimestampType, false) :: StructField("length", LongType, false) :: StructField("content", BinaryType, true) :: Nil) ``` ## How was this patch tested? To be added. Please review http://spark.apache.org/contributing.html before opening a pull request. Closes apache#24387 from WeichenXu123/binary_ds_filter. Lead-authored-by: WeichenXu <[email protected]> Co-authored-by: Xiangrui Meng <[email protected]> Signed-off-by: Xiangrui Meng <[email protected]>

init pr

83380a7

mengxr requested changes Apr 17, 2019

View reviewed changes

address comments

8acd36e

WeichenXu123 changed the title ~~[SPARK-27473][SQL][WIP] Support filter push down for status fields in binary file data source~~ [SPARK-27473][SQL] Support filter push down for status fields in binary file data source Apr 18, 2019

mengxr requested changes Apr 18, 2019

View reviewed changes

address comments

ecc156d

update test

a4aba74

update tests

06e61c7

(xiangrui) update tests (#7)

600aa9a

Change to unit test `createFilterFunction`, which should be sufficient.

WeichenXu123 and others added 3 commits April 20, 2019 11:22

nit update

7d9bdb0

add more tests

75e0a77

merge upstream

71db855

mengxr approved these changes Apr 20, 2019

View reviewed changes

asfgit closed this in 9793d9e Apr 21, 2019

WeichenXu123 deleted the binary_ds_filter branch April 22, 2019 00:10

HyukjinKwon reviewed Apr 28, 2019

View reviewed changes

[SPARK-27473][SQL] Support filter push down for status fields in binary file data source #24387

[SPARK-27473][SQL] Support filter push down for status fields in binary file data source #24387

Uh oh!

Conversation

WeichenXu123 commented Apr 17, 2019 • edited by HyukjinKwon Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented Apr 17, 2019

Uh oh!

Uh oh!

Uh oh!

Uh oh!

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!

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!

Uh oh!

mengxr commented Apr 18, 2019

Uh oh!

SparkQA commented Apr 18, 2019

Uh oh!

SparkQA commented Apr 19, 2019

Uh oh!

WeichenXu123 commented Apr 19, 2019

Uh oh!

SparkQA commented Apr 19, 2019

Uh oh!

SparkQA commented Apr 20, 2019

Uh oh!

mengxr commented Apr 20, 2019

Uh oh!

WeichenXu123 commented Apr 20, 2019

Uh oh!

mengxr commented Apr 20, 2019

Uh oh!

mengxr commented Apr 20, 2019

Uh oh!

SparkQA commented Apr 20, 2019

Uh oh!

SparkQA commented Apr 20, 2019

Uh oh!

SparkQA commented Apr 20, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

WeichenXu123 commented Apr 17, 2019 •

edited by HyukjinKwon

Loading