[SPARK-24706][SQL] ByteType and ShortType support pushdown to parquet #21682

wangyum · 2018-06-30T19:17:27Z

What changes were proposed in this pull request?

ByteType and ShortType support pushdown to parquet data source.
Benchmark result.

How was this patch tested?

unit tests

SparkQA · 2018-06-30T21:54:59Z

Test build #92503 has finished for PR 21682 at commit e9d5625.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

kiszk · 2018-07-02T02:39:09Z

retest this please

maropu · 2018-07-02T05:34:10Z

@HyukjinKwon btw, why we don't support pushdown for these types? Any historical reason?

gatorsmile · 2018-07-02T05:54:55Z

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

  private val makeEq: PartialFunction[DataType, (String, Any) => FilterPredicate] = {
    case BooleanType =>
      (n: String, v: Any) => FilterApi.eq(booleanColumn(n), v.asInstanceOf[java.lang.Boolean])
+    case ByteType =>


Let us update the comment in the line 34.

Note: For now timestamp SQL type is not supported for pushing down, because the corresponding Parquet type can be Long or INT96. There is no context about Parquet type here.

Probably INT64 or INT96 (which is long vs binary) might be a bit better while we are here.

gatorsmile · 2018-07-02T05:55:13Z

cc @gengliangwang

gatorsmile · 2018-07-02T05:56:34Z

@maropu We accidentally dropped the pushdown of these data types when we refactored the file formats. You can check the change history and find the PR.

SparkQA · 2018-07-02T06:23:50Z

Test build #92519 has finished for PR 21682 at commit e9d5625.

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

HyukjinKwon · 2018-07-02T07:01:06Z

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

  private val makeNotEq: PartialFunction[DataType, (String, Any) => FilterPredicate] = {
    case BooleanType =>
      (n: String, v: Any) => FilterApi.notEq(booleanColumn(n), v.asInstanceOf[java.lang.Boolean])
+    case ByteType =>


Just for doubly sure, is it an official way to handle short and byte? Can you double check and leave the description about the API specification (or related code bit or reference) @wangyum? Parquet one is pretty core and we should better be clear on this (so that we could also probably judge if we need a configuration or not). @rdblue too.

Usually, both byte and short would be stored as integers in Parquet. Because Parquet uses bit packing, it doesn't matter if you store them as ints (or even longs) because they'll get packed into the same space.

The important thing is to match the Parquet file's type when pushing a filter. Since Spark stores ByteType and ShortType in Parquet as INT32, this is correct.

maropu · 2018-07-02T07:03:51Z

@gatorsmile aha, thanks.

HyukjinKwon · 2018-07-02T07:05:44Z

re: #21682 (comment) - I couldn't recall related things about it. Thanks @gatorsmile. So, is it a regression? shall we update the migration guide to note that the performance effect? - like if your workload are heavily dependent on short and byte in Parquet, avoid upgrading to xx version. This optimization was restored in yy+ version. Please consider using yy+ version.

gatorsmile · 2018-07-02T07:15:17Z

It is a regression that was introduced in Spark 1.2. Almost 4 years ago. https://issues.apache.org/jira/browse/SPARK-4453

Thus, I think we do not need to document it. The pushdown can be treated as new features. : )

HyukjinKwon · 2018-07-02T07:19:19Z

Wow .. so it was 4 years ago .. okay.

gengliangwang · 2018-07-02T08:56:00Z

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

+    case ByteType =>
+      (n: String, v: Any) => FilterApi.eq(
+        intColumn(n),
+        Option(v).map(b => b.asInstanceOf[java.lang.Byte].toInt.asInstanceOf[Integer]).orNull)


I think we can use Byte instead of java.lang.Byte here.

I agree.

Also, there's no need to use Option.map because the value cannot be null. That's why the IntegerType case just casts the value.

makeEq.lift(nameToType(name)).map(_(name, null)) makes value to null.

scala> null.asInstanceOf[Short].toInt.asInstanceOf[Integer] res49: Integer = 0 scala> null.asInstanceOf[java.lang.Short].toInt.asInstanceOf[Integer] java.lang.NullPointerException at scala.Predef$.Short2short(Predef.scala:360) ... 51 elided

That's why I use Option.map here.

gengliangwang · 2018-07-02T08:56:31Z

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

+    case ShortType =>
+      (n: String, v: Any) => FilterApi.eq(
+        intColumn(n),
+        Option(v).map(b => b.asInstanceOf[java.lang.Short].toInt.asInstanceOf[Integer]).orNull)


I think we can use Short instead of java.lang.Short here.

HyukjinKwon · 2018-07-02T09:09:22Z

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

+      (n: String, v: Any) => FilterApi.eq(
+        intColumn(n),
+        Option(v).map(b => b.asInstanceOf[java.lang.Byte].toInt.asInstanceOf[Integer]).orNull)
+    case ShortType =>


Can we actually fold the cases into like:

case ByteType | ShortType | IntegerType => ... b.asInstanceOf[Number].intValue()

?

How about like this:

makeEq and makeNotEq

case ByteType | ShortType => (n: String, v: Any) => FilterApi.notEq( intColumn(n), Option(v).map(_.asInstanceOf[Number].intValue.asInstanceOf[Integer]).orNull) case IntegerType => (n: String, v: Any) => FilterApi.notEq(intColumn(n), v.asInstanceOf[Integer])

makeLt, makeLtEq, makeGt and makeGtEq:

case ByteType | ShortType => (n: String, v: Any) => FilterApi.gtEq( intColumn(n), v.asInstanceOf[Number].intValue.asInstanceOf[Integer]) case IntegerType => (n: String, v: Any) => FilterApi.gtEq(intColumn(n), v.asInstanceOf[java.lang.Integer])

rdblue · 2018-07-03T23:46:25Z

+1

I agree with some of the minor refactoring suggestions, but overall this looks correct to me.

case ShortType => -> case ByteType | ShortType =>

wangyum · 2018-07-04T02:05:17Z

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

+    case ByteType | ShortType =>
+      (n: String, v: Any) => FilterApi.eq(
+        intColumn(n),
+        Option(v).map(_.asInstanceOf[Number].intValue.asInstanceOf[Integer]).orNull)


value may be null.

wangyum · 2018-07-04T02:05:31Z

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

+    case ByteType | ShortType =>
+      (n: String, v: Any) => FilterApi.lt(
+        intColumn(n),
+        v.asInstanceOf[Number].intValue.asInstanceOf[Integer])


value cannot be null.

SparkQA · 2018-07-04T05:43:26Z

Test build #92589 has finished for PR 21682 at commit b16a607.

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

gatorsmile · 2018-07-09T16:19:56Z

ping @wangyum

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

SparkQA · 2018-07-10T04:24:08Z

Test build #92781 has finished for PR 21682 at commit 3e66530.

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

cloud-fan · 2018-07-10T05:02:21Z

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

+    case ParquetByteType | ParquetShortType | ParquetIntegerType =>
+      (n: String, v: Any) => FilterApi.eq(
+        intColumn(n),
+        Option(v).map(_.asInstanceOf[Number].intValue.asInstanceOf[Integer]).orNull)


when v can be null?

makeEq and makeNotEq may be null:

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

Lines 241 to 242 in 021145f

case sources.IsNull(name) if canMakeFilterOn(name) =>

makeEq.lift(nameToType(name)).map(_(name, null))

cloud-fan · 2018-07-10T07:58:25Z

thanks, merging to master!

ByteType and ShortType pushdown to parquet

e9d5625

gatorsmile reviewed Jul 2, 2018

View reviewed changes

HyukjinKwon reviewed Jul 2, 2018

View reviewed changes

gengliangwang reviewed Jul 2, 2018

View reviewed changes

HyukjinKwon reviewed Jul 2, 2018

View reviewed changes

case ByteType =>

b16a607

case ShortType => -> case ByteType | ShortType =>

wangyum commented Jul 4, 2018

View reviewed changes

wangyum added 2 commits July 10, 2018 08:23

Merge remote-tracking branch 'upstream/master' into SPARK-24706

d12129a

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

Update benchmark result.

3e66530

wangyum mentioned this pull request Jul 10, 2018

[SPARK-24716][SQL] Refactor ParquetFilters #21696

Closed

cloud-fan reviewed Jul 10, 2018

View reviewed changes

asfgit closed this in a289009 Jul 10, 2018

	case sources.IsNull(name) if canMakeFilterOn(name) =>
	makeEq.lift(nameToType(name)).map(_(name, null))

[SPARK-24706][SQL] ByteType and ShortType support pushdown to parquet #21682

[SPARK-24706][SQL] ByteType and ShortType support pushdown to parquet #21682

Uh oh!

Conversation

wangyum commented Jun 30, 2018 • edited 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 Jun 30, 2018

Uh oh!

kiszk commented Jul 2, 2018

Uh oh!

maropu commented Jul 2, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gatorsmile commented Jul 2, 2018

Uh oh!

gatorsmile commented Jul 2, 2018

Uh oh!

SparkQA commented Jul 2, 2018

Uh oh!

HyukjinKwon Jul 2, 2018 • 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!

maropu commented Jul 2, 2018

Uh oh!

HyukjinKwon commented Jul 2, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gatorsmile commented Jul 2, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

HyukjinKwon commented Jul 2, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wangyum Jul 4, 2018 • 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!

rdblue commented Jul 3, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 4, 2018

Uh oh!

gatorsmile commented Jul 9, 2018

Uh oh!

SparkQA commented Jul 10, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Jul 10, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

wangyum commented Jun 30, 2018 •

edited

Loading

HyukjinKwon Jul 2, 2018 •

edited

Loading

HyukjinKwon commented Jul 2, 2018 •

edited

Loading

gatorsmile commented Jul 2, 2018 •

edited

Loading

wangyum Jul 4, 2018 •

edited

Loading