Skip to content

[SPARK-60322][SQL] Fix Parquet and ORC filter pushdown dropping rows for comparisons with a zero #59574

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

Open
SEPURI-SAI-KRISHNA wants to merge 1 commit into
base: master
Choose a base branch
from
+212 −28
Open

[SPARK-60322][SQL] Fix Parquet and ORC filter pushdown dropping rows for comparisons with a zero #59574

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.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 46 additions & 13 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*
Expand Down Expand Up @@ -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()
Expand All @@ -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())

Expand Down
89 changes: 75 additions & 14 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 @@ -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 =>
Expand All @@ -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 =>
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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) =>
Expand Down Expand Up @@ -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) =>
Expand Down Expand Up @@ -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) =>
Expand Down Expand Up @@ -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) =>
Expand Down Expand Up @@ -783,15 +834,25 @@ 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)

case ParquetDoubleType =>
(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)

Expand Down
38 changes: 38 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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, _) =>
Expand Down
54 changes: 53 additions & 1 deletion ...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 @@ -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
Expand Down Expand Up @@ -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) =>
Expand Down