From 13584ee8c34412c16f0317f5154268434caee0b1 Mon Sep 17 00:00:00 2001 From: hyukjinkwon Date: Wed, 19 Apr 2017 15:24:15 +0900 Subject: [PATCH 1/6] Parquet predicate pushdown on columns with dots return empty results --- .../datasources/parquet/ParquetFilters.scala | 41 +++++++++++++++++- .../parquet/ParquetFilterSuite.scala | 42 ++++++++++++++++++- 2 files changed, 81 insertions(+), 2 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala index a6a6cef5861f3..1a780a28c77d2 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala @@ -18,7 +18,8 @@ package org.apache.spark.sql.execution.datasources.parquet import org.apache.parquet.filter2.predicate._ -import org.apache.parquet.filter2.predicate.FilterApi._ +import org.apache.parquet.filter2.predicate.Operators.{Column, SupportsEqNotEq, SupportsLtGt} +import org.apache.parquet.hadoop.metadata.ColumnPath import org.apache.parquet.io.api.Binary import org.apache.spark.sql.sources @@ -29,6 +30,8 @@ import org.apache.spark.sql.types._ */ private[parquet] object ParquetFilters { + import ParquetColumns._ + private val makeEq: PartialFunction[DataType, (String, Any) => FilterPredicate] = { case BooleanType => (n: String, v: Any) => FilterApi.eq(booleanColumn(n), v.asInstanceOf[java.lang.Boolean]) @@ -235,3 +238,39 @@ private[parquet] object ParquetFilters { } } } + +/** + * Note that, this is a hacky workaround to allow dots in column names. Currently, column APIs + * in Parquet's `FilterApi` only allows dot-separated names so here we resemble those columns + * but only allow single column path that allows dots in the names as we don't currently push + * down filters with nested fields. + */ +private[parquet] object ParquetColumns { + def intColumn(columnPath: String): Column[Integer] with SupportsLtGt = { + new Column[Integer] (ColumnPath.get(columnPath), classOf[Integer]) with SupportsLtGt + } + + def longColumn(columnPath: String): Column[java.lang.Long] with SupportsLtGt = { + new Column[java.lang.Long] ( + ColumnPath.get(columnPath), classOf[java.lang.Long]) with SupportsLtGt + } + + def floatColumn(columnPath: String): Column[java.lang.Float] with SupportsLtGt = { + new Column[java.lang.Float] ( + ColumnPath.get(columnPath), classOf[java.lang.Float]) with SupportsLtGt + } + + def doubleColumn(columnPath: String): Column[java.lang.Double] with SupportsLtGt = { + new Column[java.lang.Double] ( + ColumnPath.get(columnPath), classOf[java.lang.Double]) with SupportsLtGt + } + + def booleanColumn(columnPath: String): Column[java.lang.Boolean] with SupportsEqNotEq = { + new Column[java.lang.Boolean] ( + ColumnPath.get(columnPath), classOf[java.lang.Boolean]) with SupportsEqNotEq + } + + def binaryColumn(columnPath: String): Column[Binary] with SupportsLtGt = { + new Column[Binary] (ColumnPath.get(columnPath), classOf[Binary]) with SupportsLtGt + } +} diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala index dd53b561326f3..084930b964efc 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala @@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.datasources.parquet import java.nio.charset.StandardCharsets import org.apache.parquet.filter2.predicate.{FilterPredicate, Operators} -import org.apache.parquet.filter2.predicate.FilterApi._ +import org.apache.parquet.filter2.predicate.FilterApi.{and, gt, lt} import org.apache.parquet.filter2.predicate.Operators.{Column => _, _} import org.apache.spark.sql._ @@ -28,6 +28,7 @@ import org.apache.spark.sql.catalyst.dsl.expressions._ import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.catalyst.planning.PhysicalOperation import org.apache.spark.sql.execution.datasources.{DataSourceStrategy, HadoopFsRelation, LogicalRelation} +import org.apache.spark.sql.execution.datasources.parquet.ParquetColumns.{doubleColumn, intColumn} import org.apache.spark.sql.functions._ import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.test.SharedSQLContext @@ -538,6 +539,45 @@ class ParquetFilterSuite extends QueryTest with ParquetTest with SharedSQLContex // scalastyle:on nonascii } } + + test("SPARK-20364: Predicate pushdown for columns with a '.' in them") { + import testImplicits._ + + Seq(true, false).foreach { vectorized => + withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> vectorized.toString) { + withTempPath { path => + Seq(Some(1), None).toDF("col.dots").write.parquet(path.getAbsolutePath) + assert(spark.read.parquet(path.getAbsolutePath).where("`col.dots` > 0").count() == 1) + } + + withTempPath { path => + Seq(Some(1L), None).toDF("col.dots").write.parquet(path.getAbsolutePath) + assert(spark.read.parquet(path.getAbsolutePath).where("`col.dots` >= 1L").count() == 1) + } + + withTempPath { path => + Seq(Some(1.0F), None).toDF("col.dots").write.parquet(path.getAbsolutePath) + assert(spark.read.parquet(path.getAbsolutePath).where("`col.dots` < 2.0").count() == 1) + } + + withTempPath { path => + Seq(Some(1.0D), None).toDF("col.dots").write.parquet(path.getAbsolutePath) + assert(spark.read.parquet(path.getAbsolutePath).where("`col.dots` <= 1.0D").count() == 1) + } + + withTempPath { path => + Seq(true, false).toDF("col.dots").write.parquet(path.getAbsolutePath) + assert(spark.read.parquet(path.getAbsolutePath).where("`col.dots` == true").count() == 1) + } + + withTempPath { path => + Seq("apple", null).toDF("col.dots").write.parquet(path.getAbsolutePath) + assert( + spark.read.parquet(path.getAbsolutePath).where("`col.dots` IS NOT NULL").count() == 1) + } + } + } + } } class NumRowGroupsAcc extends AccumulatorV2[Integer, Integer] { From 05ada157ca0f1a133c9f7a931ff6267ebc18eac5 Mon Sep 17 00:00:00 2001 From: hyukjinkwon Date: Thu, 20 Apr 2017 11:20:43 +0900 Subject: [PATCH 2/6] Add negative case so that it does not push down filters for nested columns --- .../datasources/FileSourceStrategySuite.scala | 35 +++++++++++++++++-- 1 file changed, 33 insertions(+), 2 deletions(-) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala index fa3c69612704d..cd0cba036c3d8 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala @@ -487,6 +487,20 @@ class FileSourceStrategySuite extends QueryTest with SharedSQLContext with Predi } } + test("no filter puwhdown for nested field access") { + val table = createTable( + files = Seq("file1" -> 1), + format = classOf[TestFileFormatWithNestedSchema].getName) + + checkScan(table.where("a1 = 1"))(_ => ()) + // Check `a1` access pushes the predicate. + checkDataFilters(Set(IsNotNull("a1"), EqualTo("a1", 1))) + + checkScan(table.where("a2.c1 = 1"))(_ => ()) + // Check `a2.c1` access does not pushes the predicate. + checkDataFilters(Set(IsNotNull("a2"))) + } + // Helpers for checking the arguments passed to the FileFormat. protected val checkPartitionSchema = @@ -537,7 +551,8 @@ class FileSourceStrategySuite extends QueryTest with SharedSQLContext with Predi */ def createTable( files: Seq[(String, Int)], - buckets: Int = 0): DataFrame = { + buckets: Int = 0, + format: String = classOf[TestFileFormat].getName): DataFrame = { val tempDir = Utils.createTempDir() files.foreach { case (name, size) => @@ -547,7 +562,7 @@ class FileSourceStrategySuite extends QueryTest with SharedSQLContext with Predi } val df = spark.read - .format(classOf[TestFileFormat].getName) + .format(format) .load(tempDir.getCanonicalPath) if (buckets > 0) { @@ -632,6 +647,22 @@ class TestFileFormat extends TextBasedFileFormat { } } +/** + * A test [[FileFormat]] that records the arguments passed to buildReader, and returns nothing. + * Unlike the one above, this one has a nested schema. + */ +class TestFileFormatWithNestedSchema extends TestFileFormat { + override def inferSchema( + sparkSession: SparkSession, + options: Map[String, String], + files: Seq[FileStatus]): Option[StructType] = + Some(StructType(Nil) + .add("a1", IntegerType) + .add("a2", + StructType(Nil) + .add("c1", IntegerType) + .add("c2", IntegerType))) +} class LocalityTestFileSystem extends RawLocalFileSystem { private val invocations = new AtomicInteger(0) From db113a545966cbe57f5dd3690a0ddebdb1a7f793 Mon Sep 17 00:00:00 2001 From: hyukjinkwon Date: Thu, 20 Apr 2017 11:24:40 +0900 Subject: [PATCH 3/6] Fix a typo in the test --- .../sql/execution/datasources/FileSourceStrategySuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala index cd0cba036c3d8..a338544608863 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala @@ -497,7 +497,7 @@ class FileSourceStrategySuite extends QueryTest with SharedSQLContext with Predi checkDataFilters(Set(IsNotNull("a1"), EqualTo("a1", 1))) checkScan(table.where("a2.c1 = 1"))(_ => ()) - // Check `a2.c1` access does not pushes the predicate. + // Check `a2.c1` access does not push the predicate. checkDataFilters(Set(IsNotNull("a2"))) } From df65bb6cfc3eccfa44e791fcf26a0889bc23103a Mon Sep 17 00:00:00 2001 From: hyukjinkwon Date: Thu, 20 Apr 2017 11:52:05 +0900 Subject: [PATCH 4/6] Add record-by-record test too --- .../datasources/parquet/ParquetFilterSuite.scala | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala index 084930b964efc..a2820f59c789c 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala @@ -577,6 +577,16 @@ class ParquetFilterSuite extends QueryTest with ParquetTest with SharedSQLContex } } } + + withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> false.toString) { + withTempPath { path => + Seq("apple", null).toDF("col.dots").write.parquet(path.getAbsolutePath) + // This checks record-by-record filtering in Parquet's filter2. + val num = stripSparkFilter( + spark.read.parquet(path.getAbsolutePath).where("`col.dots` IS NULL")).count() + assert(num == 1) + } + } } } From 45ac9c75c69099957b3de4240d6536293ccb2250 Mon Sep 17 00:00:00 2001 From: hyukjinkwon Date: Fri, 21 Apr 2017 07:00:52 +0900 Subject: [PATCH 5/6] Fix a typo --- .../sql/execution/datasources/FileSourceStrategySuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala index a338544608863..e8a5950279b6c 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala @@ -487,7 +487,7 @@ class FileSourceStrategySuite extends QueryTest with SharedSQLContext with Predi } } - test("no filter puwhdown for nested field access") { + test("no filter pushdown for nested field access") { val table = createTable( files = Seq("file1" -> 1), format = classOf[TestFileFormatWithNestedSchema].getName) From 9f2851bfe559289f0f10f2b6fa629fb53c71c8e5 Mon Sep 17 00:00:00 2001 From: hyukjinkwon Date: Tue, 16 May 2017 08:58:13 +0900 Subject: [PATCH 6/6] Address comments here first --- .../datasources/parquet/ParquetFilters.scala | 3 +- .../parquet/ParquetFilterSuite.scala | 52 ++++++++----------- 2 files changed, 25 insertions(+), 30 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala index 1a780a28c77d2..d74c4c9e43fe0 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilters.scala @@ -243,7 +243,8 @@ private[parquet] object ParquetFilters { * Note that, this is a hacky workaround to allow dots in column names. Currently, column APIs * in Parquet's `FilterApi` only allows dot-separated names so here we resemble those columns * but only allow single column path that allows dots in the names as we don't currently push - * down filters with nested fields. + * down filters with nested fields. The functions in this object are based on + * the codes in `org.apache.parquet.filter2.predicate`. */ private[parquet] object ParquetColumns { def intColumn(columnPath: String): Column[Integer] with SupportsLtGt = { diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala index a2820f59c789c..eb0e43b9d1a48 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFilterSuite.scala @@ -545,35 +545,29 @@ class ParquetFilterSuite extends QueryTest with ParquetTest with SharedSQLContex Seq(true, false).foreach { vectorized => withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> vectorized.toString) { - withTempPath { path => - Seq(Some(1), None).toDF("col.dots").write.parquet(path.getAbsolutePath) - assert(spark.read.parquet(path.getAbsolutePath).where("`col.dots` > 0").count() == 1) - } - - withTempPath { path => - Seq(Some(1L), None).toDF("col.dots").write.parquet(path.getAbsolutePath) - assert(spark.read.parquet(path.getAbsolutePath).where("`col.dots` >= 1L").count() == 1) - } - - withTempPath { path => - Seq(Some(1.0F), None).toDF("col.dots").write.parquet(path.getAbsolutePath) - assert(spark.read.parquet(path.getAbsolutePath).where("`col.dots` < 2.0").count() == 1) - } - - withTempPath { path => - Seq(Some(1.0D), None).toDF("col.dots").write.parquet(path.getAbsolutePath) - assert(spark.read.parquet(path.getAbsolutePath).where("`col.dots` <= 1.0D").count() == 1) - } - - withTempPath { path => - Seq(true, false).toDF("col.dots").write.parquet(path.getAbsolutePath) - assert(spark.read.parquet(path.getAbsolutePath).where("`col.dots` == true").count() == 1) - } - - withTempPath { path => - Seq("apple", null).toDF("col.dots").write.parquet(path.getAbsolutePath) - assert( - spark.read.parquet(path.getAbsolutePath).where("`col.dots` IS NOT NULL").count() == 1) + val dfs = Seq( + Seq(Some(1), None).toDF("col.dots"), + Seq(Some(1L), None).toDF("col.dots"), + Seq(Some(1.0F), None).toDF("col.dots"), + Seq(Some(1.0D), None).toDF("col.dots"), + Seq(true, false).toDF("col.dots"), + Seq("apple", null).toDF("col.dots") + ) + + val predicates = Seq( + "`col.dots` > 0", + "`col.dots` >= 1L", + "`col.dots` < 2.0", + "`col.dots` <= 1.0D", + "`col.dots` == true", + "`col.dots` IS NOT NULL" + ) + + dfs.zip(predicates).foreach { case (df, predicate) => + withTempPath { path => + df.write.parquet(path.getAbsolutePath) + assert(spark.read.parquet(path.getAbsolutePath).where(predicate).count() == 1) + } } } }