dongjoon-hyun commented on a change in pull request #27824: [SPARK-31064] [SQL]
New Parquet Predicate Filter APIs with multi-part Identifier Support
URL: https://github.com/apache/spark/pull/27824#discussion_r388708471
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
##########
@@ -126,264 +126,270 @@ class ParquetFilters(
Binary.fromConstantByteArray(fixedLengthBytes, 0, numBytes)
}
- private val makeEq: PartialFunction[ParquetSchemaType, (String, Any) =>
FilterPredicate] = {
+ private val makeEq
+ : PartialFunction[ParquetSchemaType, (Array[String], Any) =>
FilterPredicate] = {
case ParquetBooleanType =>
- (n: String, v: Any) => FilterApi.eq(booleanColumn(n),
v.asInstanceOf[JBoolean])
+ (n: Array[String], v: Any) => FilterApi.eq(booleanColumn(n),
v.asInstanceOf[JBoolean])
case ParquetByteType | ParquetShortType | ParquetIntegerType =>
- (n: String, v: Any) => FilterApi.eq(
+ (n: Array[String], v: Any) => FilterApi.eq(
intColumn(n),
Option(v).map(_.asInstanceOf[Number].intValue.asInstanceOf[Integer]).orNull)
case ParquetLongType =>
- (n: String, v: Any) => FilterApi.eq(longColumn(n), v.asInstanceOf[JLong])
+ (n: Array[String], v: Any) => FilterApi.eq(longColumn(n),
v.asInstanceOf[JLong])
case ParquetFloatType =>
- (n: String, v: Any) => FilterApi.eq(floatColumn(n),
v.asInstanceOf[JFloat])
+ (n: Array[String], v: Any) => FilterApi.eq(floatColumn(n),
v.asInstanceOf[JFloat])
case ParquetDoubleType =>
- (n: String, v: Any) => FilterApi.eq(doubleColumn(n),
v.asInstanceOf[JDouble])
+ (n: Array[String], v: Any) => FilterApi.eq(doubleColumn(n),
v.asInstanceOf[JDouble])
// Binary.fromString and Binary.fromByteArray don't accept null values
case ParquetStringType =>
- (n: String, v: Any) => FilterApi.eq(
+ (n: Array[String], v: Any) => FilterApi.eq(
binaryColumn(n),
Option(v).map(s => Binary.fromString(s.asInstanceOf[String])).orNull)
case ParquetBinaryType =>
- (n: String, v: Any) => FilterApi.eq(
+ (n: Array[String], v: Any) => FilterApi.eq(
binaryColumn(n),
Option(v).map(b =>
Binary.fromReusedByteArray(v.asInstanceOf[Array[Byte]])).orNull)
case ParquetDateType if pushDownDate =>
- (n: String, v: Any) => FilterApi.eq(
+ (n: Array[String], v: Any) => FilterApi.eq(
intColumn(n),
Option(v).map(date =>
dateToDays(date.asInstanceOf[Date]).asInstanceOf[Integer]).orNull)
case ParquetTimestampMicrosType if pushDownTimestamp =>
- (n: String, v: Any) => FilterApi.eq(
+ (n: Array[String], v: Any) => FilterApi.eq(
longColumn(n),
Option(v).map(t =>
DateTimeUtils.fromJavaTimestamp(t.asInstanceOf[Timestamp])
.asInstanceOf[JLong]).orNull)
case ParquetTimestampMillisType if pushDownTimestamp =>
- (n: String, v: Any) => FilterApi.eq(
+ (n: Array[String], v: Any) => FilterApi.eq(
longColumn(n),
Option(v).map(_.asInstanceOf[Timestamp].getTime.asInstanceOf[JLong]).orNull)
case ParquetSchemaType(DECIMAL, INT32, _, _) if pushDownDecimal =>
- (n: String, v: Any) => FilterApi.eq(
+ (n: Array[String], v: Any) => FilterApi.eq(
intColumn(n),
Option(v).map(d => decimalToInt32(d.asInstanceOf[JBigDecimal])).orNull)
case ParquetSchemaType(DECIMAL, INT64, _, _) if pushDownDecimal =>
- (n: String, v: Any) => FilterApi.eq(
+ (n: Array[String], v: Any) => FilterApi.eq(
longColumn(n),
Option(v).map(d => decimalToInt64(d.asInstanceOf[JBigDecimal])).orNull)
case ParquetSchemaType(DECIMAL, FIXED_LEN_BYTE_ARRAY, length, _) if
pushDownDecimal =>
- (n: String, v: Any) => FilterApi.eq(
+ (n: Array[String], v: Any) => FilterApi.eq(
binaryColumn(n),
Option(v).map(d => decimalToByteArray(d.asInstanceOf[JBigDecimal],
length)).orNull)
}
- private val makeNotEq: PartialFunction[ParquetSchemaType, (String, Any) =>
FilterPredicate] = {
+ private val makeNotEq
+ : PartialFunction[ParquetSchemaType, (Array[String], Any) =>
FilterPredicate] = {
case ParquetBooleanType =>
- (n: String, v: Any) => FilterApi.notEq(booleanColumn(n),
v.asInstanceOf[JBoolean])
+ (n: Array[String], v: Any) => FilterApi.notEq(booleanColumn(n),
v.asInstanceOf[JBoolean])
case ParquetByteType | ParquetShortType | ParquetIntegerType =>
- (n: String, v: Any) => FilterApi.notEq(
+ (n: Array[String], v: Any) => FilterApi.notEq(
intColumn(n),
Option(v).map(_.asInstanceOf[Number].intValue.asInstanceOf[Integer]).orNull)
case ParquetLongType =>
- (n: String, v: Any) => FilterApi.notEq(longColumn(n),
v.asInstanceOf[JLong])
+ (n: Array[String], v: Any) => FilterApi.notEq(longColumn(n),
v.asInstanceOf[JLong])
case ParquetFloatType =>
- (n: String, v: Any) => FilterApi.notEq(floatColumn(n),
v.asInstanceOf[JFloat])
+ (n: Array[String], v: Any) => FilterApi.notEq(floatColumn(n),
v.asInstanceOf[JFloat])
case ParquetDoubleType =>
- (n: String, v: Any) => FilterApi.notEq(doubleColumn(n),
v.asInstanceOf[JDouble])
+ (n: Array[String], v: Any) => FilterApi.notEq(doubleColumn(n),
v.asInstanceOf[JDouble])
case ParquetStringType =>
- (n: String, v: Any) => FilterApi.notEq(
+ (n: Array[String], v: Any) => FilterApi.notEq(
binaryColumn(n),
Option(v).map(s => Binary.fromString(s.asInstanceOf[String])).orNull)
case ParquetBinaryType =>
- (n: String, v: Any) => FilterApi.notEq(
+ (n: Array[String], v: Any) => FilterApi.notEq(
binaryColumn(n),
Option(v).map(b =>
Binary.fromReusedByteArray(v.asInstanceOf[Array[Byte]])).orNull)
case ParquetDateType if pushDownDate =>
- (n: String, v: Any) => FilterApi.notEq(
+ (n: Array[String], v: Any) => FilterApi.notEq(
intColumn(n),
Option(v).map(date =>
dateToDays(date.asInstanceOf[Date]).asInstanceOf[Integer]).orNull)
case ParquetTimestampMicrosType if pushDownTimestamp =>
- (n: String, v: Any) => FilterApi.notEq(
+ (n: Array[String], v: Any) => FilterApi.notEq(
longColumn(n),
Option(v).map(t =>
DateTimeUtils.fromJavaTimestamp(t.asInstanceOf[Timestamp])
.asInstanceOf[JLong]).orNull)
case ParquetTimestampMillisType if pushDownTimestamp =>
- (n: String, v: Any) => FilterApi.notEq(
+ (n: Array[String], v: Any) => FilterApi.notEq(
longColumn(n),
Option(v).map(_.asInstanceOf[Timestamp].getTime.asInstanceOf[JLong]).orNull)
case ParquetSchemaType(DECIMAL, INT32, _, _) if pushDownDecimal =>
- (n: String, v: Any) => FilterApi.notEq(
+ (n: Array[String], v: Any) => FilterApi.notEq(
intColumn(n),
Option(v).map(d => decimalToInt32(d.asInstanceOf[JBigDecimal])).orNull)
case ParquetSchemaType(DECIMAL, INT64, _, _) if pushDownDecimal =>
- (n: String, v: Any) => FilterApi.notEq(
+ (n: Array[String], v: Any) => FilterApi.notEq(
longColumn(n),
Option(v).map(d => decimalToInt64(d.asInstanceOf[JBigDecimal])).orNull)
case ParquetSchemaType(DECIMAL, FIXED_LEN_BYTE_ARRAY, length, _) if
pushDownDecimal =>
- (n: String, v: Any) => FilterApi.notEq(
+ (n: Array[String], v: Any) => FilterApi.notEq(
binaryColumn(n),
Option(v).map(d => decimalToByteArray(d.asInstanceOf[JBigDecimal],
length)).orNull)
}
- private val makeLt: PartialFunction[ParquetSchemaType, (String, Any) =>
FilterPredicate] = {
+ private val makeLt
+ : PartialFunction[ParquetSchemaType, (Array[String], Any) =>
FilterPredicate] = {
Review comment:
ditto.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
