dongjoon-hyun commented on code in PR #59004:
URL: https://github.com/apache/spark/pull/59004#discussion_r4150313712
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala:
##########
@@ -281,9 +281,19 @@ private[sql] object OrcFilters extends OrcFiltersBase {
.isNull(dataTypeMap(name).fieldName, getType(name)).end())
case In(name, values) if dataTypeMap.contains(name) =>
Review Comment:
**1. Correctness: the comparison leaves still hit the same NPE.**
This fixes only the `In` leaf. `EqualTo`, `EqualNullSafe`, `LessThan`,
`LessThanOrEqual`, `GreaterThan`, and `GreaterThanOrEqual` (L245-L273) still
call `castLiteralValue(value, ...)` without a null check, so a null value
throws an NPE for integral, float, double, decimal, and ANSI interval columns
(L166-L181).
This is reachable in the default configuration. Since SPARK-43402,
`FileSourceScanExec` pushes filters that contain a scalar subquery and replaces
the subquery with its result at runtime, and that result is `NULL` for an empty
table:
```sql
CREATE TABLE t1(c1 INT) USING ORC;
INSERT INTO t1 VALUES (1), (2), (3);
CREATE TABLE t2(c2 INT) USING ORC;
SELECT * FROM t1 WHERE c1 > (SELECT max(c2) FROM t2);
```
Both before and after this PR, the query fails with `FAILED_READ_FILE`,
caused by a `NullPointerException` at
`OrcFilters$.castLiteralValue(OrcFilters.scala:167)`, and `=`, `!=`, `<=>`,
`<`, `<=`, and `>=` behave the same. With
`spark.sql.files.ignoreCorruptFiles=true`, `WHERE c1 = (SELECT max(c2) FROM t2)
OR c1 = 1` silently returns no rows instead of `1`. The DSv2 `OrcScan` does not
push runtime scalar-subquery filters, so only V1 and `spark.sql.orc.impl=hive`
are affected.
Parquet fixed the same scenario in SPARK-47120 with `value != null` guards.
Could you add the same guard to these leaves (`EqualNullSafe(name, null)` could
push `isNull` instead), or file a separate JIRA? An ORC version of the
SPARK-47120 test needs an integral, decimal, or double column. A `DATE` column
does not reproduce it, because a null date literal passes through
`castLiteralValue`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala:
##########
@@ -72,6 +72,35 @@ case class Person(name: String, age: Int, contacts:
Seq[Contact])
abstract class OrcQueryTest extends OrcTest {
import testImplicits._
+ test("SPARK-59605: ORC IN filter with NULL matches with pushdown on and
off") {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.sql(
+ "SELECT * FROM VALUES " +
+ "(CAST(NULL AS INT), 'null'), (1, 'a'), (2, 'b'), (3, 'c') AS t(id,
label)")
+ .write.orc(path)
+
+ Seq("true", "false").foreach { pushdown =>
+ withSQLConf(SQLConf.ORC_FILTER_PUSHDOWN_ENABLED.key -> pushdown) {
+ checkAnswer(
+ spark.read.orc(path)
+ .where("id IN (1, 3, NULL)")
+ .select("id", "label")
+ .orderBy("id", "label"),
+ Seq(Row(1, "a"), Row(3, "c")))
+
+ // NOT IN with NULL: under three-valued logic every row evaluates to
FALSE or NULL,
+ // never TRUE, so the result is empty.
+ checkAnswer(
+ spark.read.orc(path)
+ .where("id NOT IN (1, 3, NULL)")
Review Comment:
**3. Tests: this `NOT IN` check never reaches ORC.**
The idea from the earlier review is right, but
`ReplaceNullWithFalseInPredicate` (SPARK-34692) rewrites `NOT (id IN (1, 3,
NULL))` to `false`, and `PruneFilters` then replaces the scan with an empty
`LocalRelation`. So neither the ORC reader nor `OrcFilters` runs here in
`OrcV1QuerySuite`, `OrcV2QuerySuite`, or `HiveOrcQuerySuite`, with pushdown on
or off, and this block passes without the fix too. On 4.3.0 the optimized plan
is `LocalRelation <empty>`.
Excluding the rule lets the query reach the `Not(In)` pushdown that the code
comment describes:
```scala
withSQLConf(SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
ReplaceNullWithFalseInPredicate.ruleName) {
val df = spark.read.orc(path)
checkAnswer(df.where("id NOT IN (1, 3, NULL)"), Seq.empty[Row])
checkAnswer(df.where("id NOT IN (1, 3, NULL) OR label = 'b'"),
Row(2, "b"))
}
```
Without the fix, both queries fail with the NPE at `OrcFilters.scala:284` in
all three suites. With the fix, they return the expected answers. The second
query has a non-empty answer, so it would also catch a pushed `SearchArgument`
that prunes too much. `ReplaceNullWithFalseInPredicate` needs an import from
`org.apache.spark.sql.catalyst.optimizer`.
In the default configuration, this shape reaches ORC through a runtime
scalar subquery over a real table, such as `id NOT IN ((SELECT min(v) FROM
empty_table), 1)`, in V1 and in the Hive implementation.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala:
##########
@@ -72,6 +72,35 @@ case class Person(name: String, age: Int, contacts:
Seq[Contact])
abstract class OrcQueryTest extends OrcTest {
import testImplicits._
+ test("SPARK-59605: ORC IN filter with NULL matches with pushdown on and
off") {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.sql(
+ "SELECT * FROM VALUES " +
+ "(CAST(NULL AS INT), 'null'), (1, 'a'), (2, 'b'), (3, 'c') AS t(id,
label)")
+ .write.orc(path)
+
+ Seq("true", "false").foreach { pushdown =>
+ withSQLConf(SQLConf.ORC_FILTER_PUSHDOWN_ENABLED.key -> pushdown) {
+ checkAnswer(
+ spark.read.orc(path)
+ .where("id IN (1, 3, NULL)")
Review Comment:
**5. Tests: this check cannot observe the pushdown.**
`checkAnswer` compares the result after Spark re-applies the filter, and
with this data layout (two files, `{NULL, 1}` and `{2, 3}`, with
`SharedSparkSession`; one file with TestHive) the correct `SearchArgument` `IN
(1, 3)` prunes nothing. So this check catches the NPE, but not a change that
stops pushing an `IN` with a NULL. The unit test covers `OrcFilters` itself,
not the planner path.
The "SPARK-10623 Enable ORC PPD" test in this class already has a
`stripSparkFilter`-based `checkPredicate` over ten single-row files. One line
there checks both the NPE and the pruning:
```scala
checkPredicate($"a".isin(3, 5, 7, null), List(3, 5, 7).map(Row(_,
null)))
```
Without the fix it fails with the NPE, and with the fix the ORC reader
returns only the three matching rows, in all three suites.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala:
##########
@@ -730,6 +730,26 @@ class OrcFilterSuite extends OrcTest with
SharedSparkSession {
).toImmutableArraySeq).isEmpty)
}
+ test("SPARK-59605: IN filter with a NULL literal drops the NULL and pushes
the rest") {
+ import org.apache.spark.sql.sources._
+ val schema = StructType(Array(StructField("a", IntegerType, nullable =
true)))
+
+ def sarg(filter: Filter): Option[String] =
+ OrcFilters.createFilter(schema, Array(filter).toImmutableArraySeq)
+ .map(_.asInstanceOf[SearchArgumentImpl].toOldString)
+
+ // A NULL literal is inert: IN (1, 3, NULL) pushes the same SearchArgument
as IN (1, 3).
+ assert(sarg(In("a", Array[Any](1, 3, null))) === sarg(In("a",
Array[Any](1, 3))))
+ assert(sarg(In("a", Array[Any](1, 3, null))).isDefined)
+
+ // IN (NULL) has no non-null value to push, so no SearchArgument is
generated.
+ assert(sarg(In("a", Array[Any](null))).isEmpty)
+
+ // Not(IN (1, 3, NULL)) still builds and matches Not(IN (1, 3)); Not(IN
(NULL)) pushes nothing.
+ assert(sarg(Not(In("a", Array[Any](1, 3, null)))) === sarg(Not(In("a",
Array[Any](1, 3)))))
Review Comment:
**6. Tests: this assertion also passes when both sides are `None`.**
Unlike the `IN` pair (L743), there is no `isDefined` check here, so a change
that stopped converting `Not(In(...))` would still pass as `None === None`,
although the comment says it "still builds". Pinning the expected strings, as
the other tests in this suite do, covers both with one assertion each:
```scala
assert(sarg(In("a", Array[Any](1, 3, null))) === Some("leaf-0 = (IN a 1
3), expr = leaf-0"))
assert(sarg(Not(In("a", Array[Any](1, 3, null)))) ===
Some("leaf-0 = (IN a 1 3), expr = (not leaf-0)"))
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala:
##########
@@ -281,9 +281,19 @@ 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))
- Some(builder.startAnd().in(dataTypeMap(name).fieldName, getType(name),
- castedValues.map(_.asInstanceOf[AnyRef]): _*).end())
+ // SPARK-59605: a NULL in the IN list is inert (`x IN (.., NULL)`
matches the same rows as
+ // IN on the non-null values), and castLiteralValue would NPE on a
null literal. Drop NULLs;
+ // if none remain (e.g. `IN (NULL)`) there is nothing safe to push.
Under Not(In) the pushed
+ // NOT(IN(non-nulls)) is over-inclusive but safe: Spark re-applies the
full predicate.
+ val nonNullValues = values.filter(_ != null)
+ if (nonNullValues.isEmpty) {
Review Comment:
**7. Behavior change: an empty `IN` list.**
This branch also catches `In(name, Array())`, which has no NULL. Before this
PR, ORC's `BuilderImpl.in` threw `IllegalArgumentException: Can't create in
expression with no arguments`. For example, with `OptimizeIn` excluded through
`spark.sql.optimizer.excludedRules`, `spark.read.orc(path).where($"id".isin())`
failed in the task for V1 and in the optimizer for V2. Now the filter is simply
not pushed. That is an improvement, but the description says "No change for
`IN` lists without a `NULL`". Could you mention it there and add
`assert(sarg(In("a", Array.empty[Any])).isEmpty)`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala:
##########
@@ -281,9 +281,19 @@ 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))
- Some(builder.startAnd().in(dataTypeMap(name).fieldName, getType(name),
- castedValues.map(_.asInstanceOf[AnyRef]): _*).end())
+ // SPARK-59605: a NULL in the IN list is inert (`x IN (.., NULL)`
matches the same rows as
+ // IN on the non-null values), and castLiteralValue would NPE on a
null literal. Drop NULLs;
+ // if none remain (e.g. `IN (NULL)`) there is nothing safe to push.
Under Not(In) the pushed
+ // NOT(IN(non-nulls)) is over-inclusive but safe: Spark re-applies the
full predicate.
+ val nonNullValues = values.filter(_ != null)
+ if (nonNullValues.isEmpty) {
+ None
+ } else {
+ val fieldType = dataTypeMap(name).fieldType
Review Comment:
**11. Nit: the null check could move into the case guard.**
`fieldType` seems to be hoisted only for the line length under the extra
nesting; `dataTypeMap(name)` is still looked up again for `fieldName` and
inside `getType`. With the check in the guard, this branch keeps the shape of
its siblings, and an empty or all-NULL list falls through to `case _ => None`:
```scala
case In(name, values) if dataTypeMap.contains(name) && values.exists(_
!= null) =>
val castedValues =
values.filter(_ != null).map(v => castLiteralValue(v,
dataTypeMap(name).fieldType))
Some(builder.startAnd().in(dataTypeMap(name).fieldName,
getType(name),
castedValues.map(_.asInstanceOf[AnyRef]): _*).end())
```
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala:
##########
@@ -730,6 +730,26 @@ class OrcFilterSuite extends OrcTest with
SharedSparkSession {
).toImmutableArraySeq).isEmpty)
}
+ test("SPARK-59605: IN filter with a NULL literal drops the NULL and pushes
the rest") {
+ import org.apache.spark.sql.sources._
+ val schema = StructType(Array(StructField("a", IntegerType, nullable =
true)))
Review Comment:
**4. Tests: only `IntegerType` is covered.**
For string, boolean, date, timestamp, `TIMESTAMP_NTZ`, `TIME`, and the
nanosecond timestamp types, `castLiteralValue` returns the null unchanged, so
there was no NPE on the Spark side before this PR. ORC accepted the null
literal and built `leaf-0 = (IN s a null)`, then failed to compare it with the
row group statistics, logged `NullPointerException when evaluating predicate.
Skipping ORC PPD.`, and stopped pruning. On 4.3.0, an ORC file whose string
column has only `'x'` and `'y'` is skipped for `s IN ('a', 'b')` but read for
`s IN ('a', 'b', NULL)`. With `orc.sarg.to.filter=true`, `IN ('x', NULL)` on a
string, boolean, date, or `TIMESTAMP_NTZ` column even fails the query.
This PR fixes these types too, but nothing pins it; a change that dropped
NULLs only for numeric types would still pass. How about adding a string case?
```scala
val stringSchema = StructType(Array(StructField("s", StringType,
nullable = true)))
assert(OrcFilters.createFilter(stringSchema, Seq(In("s", Array[Any]("a",
null))))
.map(_.asInstanceOf[SearchArgumentImpl].toOldString) ===
Some("leaf-0 = (IN s a), expr = leaf-0"))
```
The code comment at L285 ("castLiteralValue would NPE on a null literal")
could mention this second failure mode as well.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala:
##########
@@ -72,6 +72,35 @@ case class Person(name: String, age: Int, contacts:
Seq[Contact])
abstract class OrcQueryTest extends OrcTest {
import testImplicits._
+ test("SPARK-59605: ORC IN filter with NULL matches with pushdown on and
off") {
+ withTempPath { dir =>
+ val path = dir.getCanonicalPath
+ spark.sql(
+ "SELECT * FROM VALUES " +
+ "(CAST(NULL AS INT), 'null'), (1, 'a'), (2, 'b'), (3, 'c') AS t(id,
label)")
+ .write.orc(path)
+
+ Seq("true", "false").foreach { pushdown =>
+ withSQLConf(SQLConf.ORC_FILTER_PUSHDOWN_ENABLED.key -> pushdown) {
+ checkAnswer(
+ spark.read.orc(path)
+ .where("id IN (1, 3, NULL)")
+ .select("id", "label")
+ .orderBy("id", "label"),
Review Comment:
**10. Nit: `select` and `orderBy` are not needed.**
The files have only `id` and `label`, so `.select("id", "label")` is a
no-op. `checkAnswer` already sorts both sides when the plan has no `Sort`, so
`.orderBy` only adds a range-partitioned sort (6 jobs per `checkAnswer` instead
of 2 on 4.3.0). The `NOT IN` query below uses neither. Also, the `"false"`
iteration never calls `OrcFilters`; the other pushdown tests in this file run
only with pushdown enabled.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilters.scala:
##########
@@ -281,9 +281,19 @@ 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))
- Some(builder.startAnd().in(dataTypeMap(name).fieldName, getType(name),
- castedValues.map(_.asInstanceOf[AnyRef]): _*).end())
+ // SPARK-59605: a NULL in the IN list is inert (`x IN (.., NULL)`
matches the same rows as
+ // IN on the non-null values), and castLiteralValue would NPE on a
null literal. Drop NULLs;
+ // if none remain (e.g. `IN (NULL)`) there is nothing safe to push.
Under Not(In) the pushed
Review Comment:
**8. Comment: "there is nothing safe to push".**
`x IN (NULL)` is never TRUE, with or without a surrounding `NOT`, so pushing
a constant would be safe. The actual reason is that no values remain, and ORC
rejects an empty `IN` list. Maybe "if none remain (e.g. `IN (NULL)`), there is
nothing to push". Note that the optimizer folds SQL `id IN (NULL)` before it
gets here; this branch is reached through the `Filter` API or a runtime scalar
subquery such as `id IN ((SELECT min(v) FROM empty_table))`.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcQuerySuite.scala:
##########
@@ -72,6 +72,35 @@ case class Person(name: String, age: Int, contacts:
Seq[Contact])
abstract class OrcQueryTest extends OrcTest {
import testImplicits._
+ test("SPARK-59605: ORC IN filter with NULL matches with pushdown on and
off") {
Review Comment:
**12. Nit: test placement.**
This becomes the first test of `OrcQueryTest`, before "Read/write All
Types", while the related pushdown tests come later ("SPARK-10623 Enable ORC
PPD", "SPARK-27160 ..."). Keeping it in `OrcQueryTest` is right, because
`HiveOrcQuerySuite` runs it too, but it would be easier to find next to those
tests.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
