diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala index 0e389ba99f15c..4c34f0a3d60a5 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala @@ -185,6 +185,24 @@ private[sql] object OrcFilters extends OrcFiltersBase { case _ => value } + // Spark treats 0.0 and -0.0 as equal, but ORC orders them like `Double.compare`, -0.0 before + // 0.0, in its statistics, and its bloom filters tell them apart. So a comparison with a zero is + // built in a form that keeps exactly the values that the Spark predicate keeps: an equality + // names both zeros, `lessThan` takes -0.0 as the bound and `lessThanEquals` takes 0.0. Keeping + // fewer values would drop rows, and keeping more would drop rows once the predicate is negated, + // which is how `>` and `>=` are built. + private val Zero = java.lang.Double.valueOf(0.0) + private val NegativeZero = java.lang.Double.valueOf(-0.0) + + /** Whether a value cast by `castLiteralValue` is a FLOAT or DOUBLE zero. */ + private def isZero(castedValue: Any): Boolean = castedValue match { + case d: java.lang.Double => d.doubleValue == 0.0 + case _ => false + } + + private def toBound(castedValue: Any, zero: java.lang.Double): Any = + if (isZero(castedValue)) zero else castedValue + /** * Build a SearchArgument and return the builder so far. * @@ -244,33 +262,45 @@ private[sql] object OrcFilters extends OrcFiltersBase { expression match { case EqualTo(name, value) if dataTypeMap.contains(name) => val castedValue = castLiteralValue(value, dataTypeMap(name).fieldType) - Some(builder.startAnd() - .equals(dataTypeMap(name).fieldName, getType(name), castedValue).end()) + if (isZero(castedValue)) { + Some(builder.startOr() + .equals(dataTypeMap(name).fieldName, getType(name), Zero) + .equals(dataTypeMap(name).fieldName, getType(name), NegativeZero).end()) + } else { + Some(builder.startAnd() + .equals(dataTypeMap(name).fieldName, getType(name), castedValue).end()) + } case EqualNullSafe(name, value) if dataTypeMap.contains(name) => val castedValue = castLiteralValue(value, dataTypeMap(name).fieldType) - Some(builder.startAnd() - .nullSafeEquals(dataTypeMap(name).fieldName, getType(name), castedValue).end()) + if (isZero(castedValue)) { + Some(builder.startOr() + .nullSafeEquals(dataTypeMap(name).fieldName, getType(name), Zero) + .nullSafeEquals(dataTypeMap(name).fieldName, getType(name), NegativeZero).end()) + } else { + Some(builder.startAnd() + .nullSafeEquals(dataTypeMap(name).fieldName, getType(name), castedValue).end()) + } case LessThan(name, value) if dataTypeMap.contains(name) => val castedValue = castLiteralValue(value, dataTypeMap(name).fieldType) - Some(builder.startAnd() - .lessThan(dataTypeMap(name).fieldName, getType(name), castedValue).end()) + Some(builder.startAnd().lessThan( + dataTypeMap(name).fieldName, getType(name), toBound(castedValue, NegativeZero)).end()) case LessThanOrEqual(name, value) if dataTypeMap.contains(name) => val castedValue = castLiteralValue(value, dataTypeMap(name).fieldType) - Some(builder.startAnd() - .lessThanEquals(dataTypeMap(name).fieldName, getType(name), castedValue).end()) + Some(builder.startAnd().lessThanEquals( + dataTypeMap(name).fieldName, getType(name), toBound(castedValue, Zero)).end()) case GreaterThan(name, value) if dataTypeMap.contains(name) => val castedValue = castLiteralValue(value, dataTypeMap(name).fieldType) - Some(builder.startNot() - .lessThanEquals(dataTypeMap(name).fieldName, getType(name), castedValue).end()) + Some(builder.startNot().lessThanEquals( + dataTypeMap(name).fieldName, getType(name), toBound(castedValue, Zero)).end()) case GreaterThanOrEqual(name, value) if dataTypeMap.contains(name) => val castedValue = castLiteralValue(value, dataTypeMap(name).fieldType) - Some(builder.startNot() - .lessThan(dataTypeMap(name).fieldName, getType(name), castedValue).end()) + Some(builder.startNot().lessThan( + dataTypeMap(name).fieldName, getType(name), toBound(castedValue, NegativeZero)).end()) case IsNull(name) if dataTypeMap.contains(name) => Some(builder.startAnd() @@ -281,7 +311,10 @@ private[sql] object OrcFilters extends OrcFiltersBase { .isNull(dataTypeMap(name).fieldName, getType(name)).end()) case In(name, values) if dataTypeMap.contains(name) => - val castedValues = values.map(v => castLiteralValue(v, dataTypeMap(name).fieldType)) + val castedValues = values.flatMap { v => + val castedValue = castLiteralValue(v, dataTypeMap(name).fieldType) + if (isZero(castedValue)) Seq(Zero, NegativeZero) else Seq(castedValue) + } Some(builder.startAnd().in(dataTypeMap(name).fieldName, getType(name), castedValues.map(_.asInstanceOf[AnyRef]): _*).end()) 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 93f4000fa2760..e97345abade33 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 @@ -501,6 +501,29 @@ class ParquetFilters( case l => l.asInstanceOf[JLong] } + // Spark treats 0.0 and -0.0 as equal, but Parquet orders them like `Double.compare`, -0.0 + // before 0.0, in its statistics and column indexes, and its dictionaries and bloom filters tell + // them apart. So a comparison with a zero is pushed down in a form that keeps exactly the values + // that the Spark predicate keeps: an equality names both zeros, and a range takes the zero on + // its side as the bound, -0.0 for `<` and `>=` and 0.0 for `<=` and `>`. Keeping fewer values + // would drop rows, and keeping more would drop rows once the predicate is negated by `Not`. + private val FloatZero = JFloat.valueOf(0.0f) + private val FloatNegativeZero = JFloat.valueOf(-0.0f) + private val DoubleZero = JDouble.valueOf(0.0) + private val DoubleNegativeZero = JDouble.valueOf(-0.0) + + private def isZero(v: Any): Boolean = v match { + case f: JFloat => f.floatValue == 0.0f + case d: JDouble => d.doubleValue == 0.0 + case _ => false + } + + private def toFloatBound(v: Any, zero: JFloat): JFloat = + if (isZero(v)) zero else v.asInstanceOf[JFloat] + + private def toDoubleBound(v: Any, zero: JDouble): JDouble = + if (isZero(v)) zero else v.asInstanceOf[JDouble] + private val makeEq: PartialFunction[ParquetSchemaType, (Array[String], Any) => FilterPredicate] = { case ParquetBooleanType => @@ -510,9 +533,19 @@ class ParquetFilters( case ParquetLongType => (n: Array[String], v: Any) => FilterApi.eq(longColumn(n), toLongValue(v)) case ParquetFloatType => - (n: Array[String], v: Any) => FilterApi.eq(floatColumn(n), v.asInstanceOf[JFloat]) + (n: Array[String], v: Any) => if (isZero(v)) { + FilterApi.or(FilterApi.eq(floatColumn(n), FloatZero), + FilterApi.eq(floatColumn(n), FloatNegativeZero)) + } else { + FilterApi.eq(floatColumn(n), v.asInstanceOf[JFloat]) + } case ParquetDoubleType => - (n: Array[String], v: Any) => FilterApi.eq(doubleColumn(n), v.asInstanceOf[JDouble]) + (n: Array[String], v: Any) => if (isZero(v)) { + FilterApi.or(FilterApi.eq(doubleColumn(n), DoubleZero), + FilterApi.eq(doubleColumn(n), DoubleNegativeZero)) + } else { + FilterApi.eq(doubleColumn(n), v.asInstanceOf[JDouble]) + } // Binary.fromString and Binary.fromByteArray don't accept null values case ParquetStringType => @@ -562,9 +595,19 @@ class ParquetFilters( case ParquetLongType => (n: Array[String], v: Any) => FilterApi.notEq(longColumn(n), toLongValue(v)) case ParquetFloatType => - (n: Array[String], v: Any) => FilterApi.notEq(floatColumn(n), v.asInstanceOf[JFloat]) + (n: Array[String], v: Any) => if (isZero(v)) { + FilterApi.and(FilterApi.notEq(floatColumn(n), FloatZero), + FilterApi.notEq(floatColumn(n), FloatNegativeZero)) + } else { + FilterApi.notEq(floatColumn(n), v.asInstanceOf[JFloat]) + } case ParquetDoubleType => - (n: Array[String], v: Any) => FilterApi.notEq(doubleColumn(n), v.asInstanceOf[JDouble]) + (n: Array[String], v: Any) => if (isZero(v)) { + FilterApi.and(FilterApi.notEq(doubleColumn(n), DoubleZero), + FilterApi.notEq(doubleColumn(n), DoubleNegativeZero)) + } else { + FilterApi.notEq(doubleColumn(n), v.asInstanceOf[JDouble]) + } case ParquetStringType => (n: Array[String], v: Any) => FilterApi.notEq( @@ -611,9 +654,11 @@ class ParquetFilters( case ParquetLongType => (n: Array[String], v: Any) => FilterApi.lt(longColumn(n), toLongValue(v)) case ParquetFloatType => - (n: Array[String], v: Any) => FilterApi.lt(floatColumn(n), v.asInstanceOf[JFloat]) + (n: Array[String], v: Any) => + FilterApi.lt(floatColumn(n), toFloatBound(v, FloatNegativeZero)) case ParquetDoubleType => - (n: Array[String], v: Any) => FilterApi.lt(doubleColumn(n), v.asInstanceOf[JDouble]) + (n: Array[String], v: Any) => + FilterApi.lt(doubleColumn(n), toDoubleBound(v, DoubleNegativeZero)) case ParquetStringType => (n: Array[String], v: Any) => @@ -650,9 +695,11 @@ class ParquetFilters( case ParquetLongType => (n: Array[String], v: Any) => FilterApi.ltEq(longColumn(n), toLongValue(v)) case ParquetFloatType => - (n: Array[String], v: Any) => FilterApi.ltEq(floatColumn(n), v.asInstanceOf[JFloat]) + (n: Array[String], v: Any) => + FilterApi.ltEq(floatColumn(n), toFloatBound(v, FloatZero)) case ParquetDoubleType => - (n: Array[String], v: Any) => FilterApi.ltEq(doubleColumn(n), v.asInstanceOf[JDouble]) + (n: Array[String], v: Any) => + FilterApi.ltEq(doubleColumn(n), toDoubleBound(v, DoubleZero)) case ParquetStringType => (n: Array[String], v: Any) => @@ -689,9 +736,11 @@ class ParquetFilters( case ParquetLongType => (n: Array[String], v: Any) => FilterApi.gt(longColumn(n), toLongValue(v)) case ParquetFloatType => - (n: Array[String], v: Any) => FilterApi.gt(floatColumn(n), v.asInstanceOf[JFloat]) + (n: Array[String], v: Any) => + FilterApi.gt(floatColumn(n), toFloatBound(v, FloatZero)) case ParquetDoubleType => - (n: Array[String], v: Any) => FilterApi.gt(doubleColumn(n), v.asInstanceOf[JDouble]) + (n: Array[String], v: Any) => + FilterApi.gt(doubleColumn(n), toDoubleBound(v, DoubleZero)) case ParquetStringType => (n: Array[String], v: Any) => @@ -728,9 +777,11 @@ class ParquetFilters( case ParquetLongType => (n: Array[String], v: Any) => FilterApi.gtEq(longColumn(n), toLongValue(v)) case ParquetFloatType => - (n: Array[String], v: Any) => FilterApi.gtEq(floatColumn(n), v.asInstanceOf[JFloat]) + (n: Array[String], v: Any) => + FilterApi.gtEq(floatColumn(n), toFloatBound(v, FloatNegativeZero)) case ParquetDoubleType => - (n: Array[String], v: Any) => FilterApi.gtEq(doubleColumn(n), v.asInstanceOf[JDouble]) + (n: Array[String], v: Any) => + FilterApi.gtEq(doubleColumn(n), toDoubleBound(v, DoubleNegativeZero)) case ParquetStringType => (n: Array[String], v: Any) => @@ -783,7 +834,12 @@ class ParquetFilters( (n: Array[String], values: Array[Any]) => val set = new HashSet[JFloat]() for (value <- values) { - set.add(value.asInstanceOf[JFloat]) + if (isZero(value)) { + set.add(FloatZero) + set.add(FloatNegativeZero) + } else { + set.add(value.asInstanceOf[JFloat]) + } } FilterApi.in(floatColumn(n), set) @@ -791,7 +847,12 @@ class ParquetFilters( (n: Array[String], values: Array[Any]) => val set = new HashSet[JDouble]() for (value <- values) { - set.add(value.asInstanceOf[JDouble]) + if (isZero(value)) { + set.add(DoubleZero) + set.add(DoubleNegativeZero) + } else { + set.add(value.asInstanceOf[JDouble]) + } } FilterApi.in(doubleColumn(n), set) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala index 6481921580e74..c9036481004e4 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala @@ -31,6 +31,7 @@ import org.apache.spark.{SparkConf, SparkException, SparkRuntimeException} import org.apache.spark.sql.{AnalysisException, Column, DataFrame, Row} import org.apache.spark.sql.catalyst.dsl.expressions._ import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.catalyst.optimizer.BooleanSimplification import org.apache.spark.sql.catalyst.planning.PhysicalOperation import org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPrecision import org.apache.spark.sql.execution.datasources.v2.ExtractV2Scan @@ -203,6 +204,43 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession { } } + test("SPARK-60322: filter pushdown - float and double zeros") { + // Spark treats 0.0 and -0.0 as equal, but ORC orders -0.0 before 0.0, and its bloom filters + // tell them apart. Each file holds zeros of one sign only, so a predicate pushed down with the + // other zero must not skip it. + Seq("FLOAT", "DOUBLE").foreach { t => + withTempPath { dir => + val path = dir.getCanonicalPath + Seq(0 -> "0.0D", 100 -> "-0.0D", 200 -> "1.0D").foreach { case (start, value) => + spark.range(start, start + 100, 1, 1) + .selectExpr("id", s"CAST($value AS $t) AS f") + .write.mode("append") + .option("orc.bloom.filter.columns", "f") + .orc(path) + } + val expected = spark.range(0, 300) + .selectExpr("id", s"CAST(CASE WHEN id < 100 THEN 0.0D WHEN id < 200 THEN -0.0D " + + s"ELSE 1.0D END AS $t) AS f") + val predicates = Seq("0.0D", "-0.0D").flatMap { zero => + val z = s"CAST($zero AS $t)" + Seq(s"f = $z", s"f <=> $z", s"f IN ($z, CAST(5.0D AS $t))", s"f != $z", + s"f < $z", s"f <= $z", s"f > $z", s"f >= $z", + s"NOT (f = $z)", s"NOT (f <=> $z)", s"NOT (f != $z)", s"NOT (f < $z)", + s"NOT (f <= $z)", s"NOT (f > $z)", s"NOT (f >= $z)") + } + withSQLConf( + SQLConf.ORC_FILTER_PUSHDOWN_ENABLED.key -> "true", + // Keeps `NOT (f < 0.0)` from being rewritten into `f >= 0.0`, so that it is pushed + // down as a `Not`. + SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> BooleanSimplification.ruleName) { + predicates.foreach { p => + checkAnswer(spark.read.orc(path).where(p), expected.where(p)) + } + } + } + } + } + test("filter pushdown - string") { withNestedOrcDataFrame( (1 to 4).map(i => Tuple1(i.toString))) { case (inputDF, colName, _) => 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 cbcb1c9843f0f..17054cafa8eae 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 @@ -43,7 +43,7 @@ import org.apache.spark.{SparkConf, SparkException, SparkRuntimeException} import org.apache.spark.sql._ import org.apache.spark.sql.catalyst.dsl.expressions._ import org.apache.spark.sql.catalyst.expressions._ -import org.apache.spark.sql.catalyst.optimizer.InferFiltersFromConstraints +import org.apache.spark.sql.catalyst.optimizer.{BooleanSimplification, InferFiltersFromConstraints} import org.apache.spark.sql.catalyst.planning.PhysicalOperation import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.parseColumnPath @@ -593,6 +593,58 @@ abstract class ParquetFilterSuite extends ParquetTest with SharedSparkSession { } } + test("SPARK-60322: filter pushdown - float and double zeros") { + // Spark treats 0.0 and -0.0 as equal, but Parquet orders -0.0 before 0.0, and its dictionaries + // and bloom filters tell them apart. Each file holds zeros of one sign only, so a predicate + // pushed down with the other zero must not skip it. + Seq("FLOAT", "DOUBLE").foreach { t => + // Parquet writes a bloom filter only for a column that is not dictionary encoded. + Seq(true, false).foreach { dictionary => + withTempPath { dir => + val path = dir.getCanonicalPath + Seq(0 -> "0.0D", 100 -> "-0.0D", 200 -> "1.0D").foreach { case (start, value) => + spark.range(start, start + 100, 1, 1) + .selectExpr("id", s"CAST($value AS $t) AS f") + .write.mode("append") + .option(ParquetOutputFormat.ENABLE_DICTIONARY, dictionary.toString) + .option(ParquetOutputFormat.BLOOM_FILTER_ENABLED + "#f", (!dictionary).toString) + .parquet(path) + } + dir.listFiles().filter(_.getName.endsWith(".parquet")).foreach { file => + val in = HadoopInputFile.fromPath( + new Path(file.getCanonicalPath), spark.sessionState.newHadoopConf()) + Utils.tryWithResource(ParquetFileReader.open(in)) { reader => + val chunk = reader.getFooter.getBlocks.get(0).getColumns.get(1) + assert(chunk.hasDictionaryPage === dictionary) + assert((chunk.getBloomFilterOffset >= 0) === !dictionary) + } + } + val expected = spark.range(0, 300) + .selectExpr("id", s"CAST(CASE WHEN id < 100 THEN 0.0D WHEN id < 200 THEN -0.0D " + + s"ELSE 1.0D END AS $t) AS f") + val predicates = Seq("0.0D", "-0.0D").flatMap { zero => + val z = s"CAST($zero AS $t)" + val many = (1 to 12).map(i => s"CAST($i.5D AS $t)").mkString(", ") + Seq(s"f = $z", s"f <=> $z", s"f IN ($z, CAST(5.0D AS $t))", s"f IN ($z, $many)", + s"f != $z", s"f < $z", s"f <= $z", s"f > $z", s"f >= $z", + s"NOT (f = $z)", s"NOT (f != $z)", s"NOT (f < $z)", s"NOT (f <= $z)", + s"NOT (f > $z)", s"NOT (f >= $z)") + } + withSQLConf( + // Keeps `NOT (f < 0.0)` from being rewritten into `f >= 0.0`, so that it is pushed + // down as a `Not`. + SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> BooleanSimplification.ruleName, + // The non-vectorized reader also applies the record-level filter of this suite. + SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false") { + predicates.foreach { p => + checkAnswer(spark.read.parquet(path).where(p), expected.where(p)) + } + } + } + } + } + } + test("filter pushdown - string") { val data = (1 to 4).map(i => Tuple1(Option(i.toString))) withNestedParquetDataFrame(data) { case (inputDF, colName, resultFun) =>