Skip to content
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

Jump to bottom

[SPARK-39393][SQL] Parquet data source only supports push-down predicate filters for non-repeated primitive types #36781

Closed
wants to merge 1 commit into from
Closed

[SPARK-39393][SQL] Parquet data source only supports push-down predicate filters for non-repeated primitive types #36781

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
6 changes: 5 additions & 1 deletion ...re/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala
View file
Open in desktop
Expand Up @@ -33,6 +33,7 @@ import org.apache.parquet.schema.{GroupType, LogicalTypeAnnotation, MessageType,
import org.apache.parquet.schema.LogicalTypeAnnotation.{DecimalLogicalTypeAnnotation, TimeUnit}
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName._
import org.apache.parquet.schema.Type.Repetition

import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils, IntervalUtils}
import org.apache.spark.sql.catalyst.util.RebaseDateTime.{rebaseGregorianToJulianDays, rebaseGregorianToJulianMicros, RebaseSpec}
Expand Down Expand Up @@ -64,7 +65,10 @@ class ParquetFilters(
fields: Seq[Type],
parentFieldNames: Array[String] = Array.empty): Seq[ParquetPrimitiveField] = {
fields.flatMap {
case p: PrimitiveType =>
// Parquet only supports predicate push-down for non-repeated primitive types.
// TODO(SPARK-39393): Remove extra condition when parquet added filter predicate support for
// repeated columns (https://issues.apache.org/jira/browse/PARQUET-34)
case p: PrimitiveType if p.getRepetition != Repetition.REPEATED =>
Some(ParquetPrimitiveField(fieldNames = parentFieldNames :+ p.getName,
fieldType = ParquetSchemaType(p.getLogicalTypeAnnotation,
p.getPrimitiveTypeName, p.getTypeLength)))
Expand Down
29 changes: 29 additions & 0 deletions ...rc/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala
View file
Open in desktop
Expand Up @@ -17,6 +17,7 @@

package org.apache.spark.sql.execution.datasources.parquet

import java.io.File
import java.math.{BigDecimal => JBigDecimal}
import java.nio.charset.StandardCharsets
import java.sql.{Date, Timestamp}
Expand Down Expand Up @@ -1316,6 +1317,34 @@ abstract class ParquetFilterSuite extends QueryTest with ParquetTest with Shared
}
}

test("SPARK-39393: Do not push down predicate filters for repeated primitive fields") {
import ParquetCompatibilityTest._
withTempDir { dir =>
val protobufParquetFilePath = new File(dir, "protobuf-parquet").getCanonicalPath

val protobufSchema =
"""message protobuf_style {
| repeated int32 f;
|}
""".stripMargin

writeDirect(protobufParquetFilePath, protobufSchema, { rc =>
rc.message {
rc.field("f", 0) {
rc.addInteger(1)
rc.addInteger(2)
}
}
})

// If the "isnotnull(f)" filter gets pushed down, this query will throw an exception
// since column "f" is repeated primitive column in the Parquet file.
checkAnswer(
spark.read.parquet(dir.getCanonicalPath).filter("isnotnull(f)"),
Seq(Row(Seq(1, 2))))
}
}

test("Filters should be pushed down for vectorized Parquet reader at row group level") {
import testImplicits._

Expand Down