Skip to content

[SPARK-24549][SQL] Support Decimal type push down to the parquet data sources #21556

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

[SPARK-24549][SQL] Support Decimal type push down to the parquet data sources #21556

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
9832661
32BitDecimalType and 64BitDecimalType support push down to the data s…
wangyum Jun 13, 2018
51d8540
Add PARQUET_FILTER_PUSHDOWN_DECIMAL_ENABLED conf
wangyum Jun 15, 2018
0b5d0e7
Fully support decimal push-down.
wangyum Jun 28, 2018
e3f0736
Merge remote-tracking branch 'upstream/master' into SPARK-24549
wangyum Jul 4, 2018
f160648
Fix compile errors
wangyum Jul 4, 2018
c7308ab
Merge Legacy DecimalType and ByteArrayDecimalType
wangyum Jul 5, 2018
4eaa308
Merge remote-tracking branch 'upstream/master' into SPARK-24549
wangyum Jul 6, 2018
341fe59
Add validate check and Update benchmark
wangyum Jul 6, 2018
7128539
Remove duplicate ParquetBinaryType
wangyum Jul 7, 2018
b103cb5
Merge remote-tracking branch 'upstream/master' into SPARK-24549
wangyum Jul 11, 2018
16528f3
Refactor test
wangyum Jul 11, 2018
33d1f18
Minor update
wangyum Jul 12, 2018
f73eab2
Fix decimal(9, 2) benchmark out of range
wangyum Jul 12, 2018
e713698
Improvement comment
wangyum Jul 14, 2018
e31c201
Merge remote-tracking branch 'upstream/master' into SPARK-24549
wangyum Jul 15, 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
60 changes: 60 additions & 0 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 @@ -62,6 +62,16 @@ private[parquet] class ParquetFilters(pushDownDate: Boolean) {
(n: String, v: Any) => FilterApi.eq(
intColumn(n),
Option(v).map(date => dateToDays(date.asInstanceOf[Date]).asInstanceOf[Integer]).orNull)
case decimal: DecimalType if DecimalType.is32BitDecimalType(decimal) =>
(n: String, v: Any) => FilterApi.eq(
intColumn(n),
Option(v).map(_.asInstanceOf[java.math.BigDecimal].unscaledValue().intValue()

@wangyum wangyum Jun 14, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

REF:

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

Line 219 in 21a7bfd

val unscaledLong = row.getDecimal(ordinal, precision, scale).toUnscaledLong

.asInstanceOf[Integer]).orNull)
case decimal: DecimalType if DecimalType.is64BitDecimalType(decimal) =>
(n: String, v: Any) => FilterApi.eq(
longColumn(n),
Option(v).map(_.asInstanceOf[java.math.BigDecimal].unscaledValue().longValue()
.asInstanceOf[java.lang.Long]).orNull)
}

private val makeNotEq: PartialFunction[DataType, (String, Any) => FilterPredicate] = {
Expand All @@ -88,6 +98,16 @@ private[parquet] class ParquetFilters(pushDownDate: Boolean) {
(n: String, v: Any) => FilterApi.notEq(
intColumn(n),
Option(v).map(date => dateToDays(date.asInstanceOf[Date]).asInstanceOf[Integer]).orNull)
case decimal: DecimalType if DecimalType.is32BitDecimalType(decimal) =>
(n: String, v: Any) => FilterApi.notEq(
intColumn(n),
Option(v).map(_.asInstanceOf[java.math.BigDecimal].unscaledValue().intValue()
.asInstanceOf[Integer]).orNull)
case decimal: DecimalType if DecimalType.is64BitDecimalType(decimal) =>
(n: String, v: Any) => FilterApi.notEq(
longColumn(n),
Option(v).map(_.asInstanceOf[java.math.BigDecimal].unscaledValue().longValue()
.asInstanceOf[java.lang.Long]).orNull)
}

private val makeLt: PartialFunction[DataType, (String, Any) => FilterPredicate] = {
Expand All @@ -111,6 +131,16 @@ private[parquet] class ParquetFilters(pushDownDate: Boolean) {
(n: String, v: Any) => FilterApi.lt(
intColumn(n),
Option(v).map(date => dateToDays(date.asInstanceOf[Date]).asInstanceOf[Integer]).orNull)
case decimal: DecimalType if DecimalType.is32BitDecimalType(decimal) =>
(n: String, v: Any) => FilterApi.lt(
intColumn(n),
Option(v).map(_.asInstanceOf[java.math.BigDecimal].unscaledValue().intValue()
.asInstanceOf[Integer]).orNull)
case decimal: DecimalType if DecimalType.is64BitDecimalType(decimal) =>
(n: String, v: Any) => FilterApi.lt(
longColumn(n),
Option(v).map(_.asInstanceOf[java.math.BigDecimal].unscaledValue().longValue()
.asInstanceOf[java.lang.Long]).orNull)
}

private val makeLtEq: PartialFunction[DataType, (String, Any) => FilterPredicate] = {
Expand All @@ -134,6 +164,16 @@ private[parquet] class ParquetFilters(pushDownDate: Boolean) {
(n: String, v: Any) => FilterApi.ltEq(
intColumn(n),
Option(v).map(date => dateToDays(date.asInstanceOf[Date]).asInstanceOf[Integer]).orNull)
case decimal: DecimalType if DecimalType.is32BitDecimalType(decimal) =>
(n: String, v: Any) => FilterApi.ltEq(
intColumn(n),
Option(v).map(_.asInstanceOf[java.math.BigDecimal].unscaledValue().intValue()
.asInstanceOf[Integer]).orNull)
case decimal: DecimalType if DecimalType.is64BitDecimalType(decimal) =>
(n: String, v: Any) => FilterApi.ltEq(
longColumn(n),
Option(v).map(_.asInstanceOf[java.math.BigDecimal].unscaledValue().longValue()
.asInstanceOf[java.lang.Long]).orNull)
}

private val makeGt: PartialFunction[DataType, (String, Any) => FilterPredicate] = {
Expand All @@ -157,6 +197,16 @@ private[parquet] class ParquetFilters(pushDownDate: Boolean) {
(n: String, v: Any) => FilterApi.gt(
intColumn(n),
Option(v).map(date => dateToDays(date.asInstanceOf[Date]).asInstanceOf[Integer]).orNull)
case decimal: DecimalType if DecimalType.is32BitDecimalType(decimal) =>
(n: String, v: Any) => FilterApi.gt(
intColumn(n),
Option(v).map(_.asInstanceOf[java.math.BigDecimal].unscaledValue().intValue()
.asInstanceOf[Integer]).orNull)
case decimal: DecimalType if DecimalType.is64BitDecimalType(decimal) =>
(n: String, v: Any) => FilterApi.gt(
longColumn(n),
Option(v).map(_.asInstanceOf[java.math.BigDecimal].unscaledValue().longValue()
.asInstanceOf[java.lang.Long]).orNull)
}

private val makeGtEq: PartialFunction[DataType, (String, Any) => FilterPredicate] = {
Expand All @@ -180,6 +230,16 @@ private[parquet] class ParquetFilters(pushDownDate: Boolean) {
(n: String, v: Any) => FilterApi.gtEq(
intColumn(n),
Option(v).map(date => dateToDays(date.asInstanceOf[Date]).asInstanceOf[Integer]).orNull)
case decimal: DecimalType if DecimalType.is32BitDecimalType(decimal) =>
(n: String, v: Any) => FilterApi.gtEq(
intColumn(n),
Option(v).map(_.asInstanceOf[java.math.BigDecimal].unscaledValue().intValue()
.asInstanceOf[Integer]).orNull)
case decimal: DecimalType if DecimalType.is64BitDecimalType(decimal) =>
(n: String, v: Any) => FilterApi.gtEq(
longColumn(n),
Option(v).map(_.asInstanceOf[java.math.BigDecimal].unscaledValue().longValue()
.asInstanceOf[java.lang.Long]).orNull)
}

/**
Expand Down
40 changes: 40 additions & 0 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 @@ -359,6 +359,46 @@ class ParquetFilterSuite extends QueryTest with ParquetTest with SharedSQLContex
}
}

test("filter pushdown - decimal(is32BitDecimalType & is64BitDecimalType)") {
def testDecimalPushDown(data: DataFrame)(f: DataFrame => Unit): Unit = {
withTempPath { file =>
data.write.parquet(file.getCanonicalPath)
readParquetFile(file.toString, true)(f)
}
}

Seq(s"_1 decimal(${Decimal.MAX_INT_DIGITS}, 2)", s"_1 decimal(${Decimal.MAX_LONG_DIGITS}, 2)")
.foreach { schemaDDL =>
val schema = StructType.fromDDL(schemaDDL)
val rdd = spark.sparkContext.parallelize((1 to 4).map(i => Row(new java.math.BigDecimal(i))))
val dataFrame = spark.createDataFrame(rdd, schema)
testDecimalPushDown(dataFrame) { implicit df =>
assert(df.schema === schema)
checkFilterPredicate('_1.isNull, classOf[Eq[_]], Seq.empty[Row])
checkFilterPredicate('_1.isNotNull, classOf[NotEq[_]], (1 to 4).map(Row.apply(_)))

checkFilterPredicate('_1 === 1, classOf[Eq[_]], 1)
checkFilterPredicate('_1 <=> 1, classOf[Eq[_]], 1)
checkFilterPredicate('_1 =!= 1, classOf[NotEq[_]], (2 to 4).map(Row.apply(_)))

checkFilterPredicate('_1 < 2, classOf[Lt[_]], 1)
checkFilterPredicate('_1 > 3, classOf[Gt[_]], 4)
checkFilterPredicate('_1 <= 1, classOf[LtEq[_]], 1)
checkFilterPredicate('_1 >= 4, classOf[GtEq[_]], 4)

checkFilterPredicate(Literal(1) === '_1, classOf[Eq[_]], 1)
checkFilterPredicate(Literal(1) <=> '_1, classOf[Eq[_]], 1)
checkFilterPredicate(Literal(2) > '_1, classOf[Lt[_]], 1)
checkFilterPredicate(Literal(3) < '_1, classOf[Gt[_]], 4)
checkFilterPredicate(Literal(1) >= '_1, classOf[LtEq[_]], 1)
checkFilterPredicate(Literal(4) <= '_1, classOf[GtEq[_]], 4)

checkFilterPredicate(!('_1 < 4), classOf[GtEq[_]], 4)
checkFilterPredicate('_1 < 2 || '_1 > 3, classOf[Operators.Or], Seq(Row(1), Row(4)))
}
}
}

test("SPARK-6554: don't push down predicates which reference partition columns") {
import testImplicits._

Expand Down