This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 57f0b81e9228 fix(spark): don't prune files with unknown null-count on
IS NULL (#19646)
57f0b81e9228 is described below
commit 57f0b81e922887fe5e4ec50aaee65840f884b417
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Fri Aug 21 07:43:31 2026 +0530
fix(spark): don't prune files with unknown null-count on IS NULL (#19646)
* fix(spark): don't prune files with unknown null-count on IS NULL
`colA is null` translated to the column-stats predicate `colA_nullCount >
0`,
and `colA <=> null` to `colA_nullCount = null`. The transposed column stats
index carries a null null-count whenever a file has no stats record for the
column (for ex. a column added by schema evolution, or a type the index does
not support), so both predicates evaluate to null for those rows and the
files
end up pruned -- even though all of their records are null for that column,
i.e. exactly the rows the query asks for.
Both arms now translate to `colA_nullCount is null OR colA_nullCount > 0`,
keeping files whose null-count is unknown and leaving the pruning decision
unchanged when the null-count is known. This mirrors the unknown-stats
handling
the `is not null` translation already had.
* address review feedback: spell null-count comments as IS NULL
---
.../apache/spark/sql/hudi/DataSkippingUtils.scala | 39 ++++++++++++++--------
.../org/apache/hudi/TestDataSkippingUtils.scala | 29 ++++++++++++++++
2 files changed, 55 insertions(+), 13 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/DataSkippingUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/DataSkippingUtils.scala
index a15df36e9ce1..c2cae5a98b87 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/DataSkippingUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/DataSkippingUtils.scala
@@ -160,11 +160,12 @@ object DataSkippingUtils extends Logging {
Option.empty
})
- // Filter "colA = null"
- // Translates to "colA_nullCount = null" for index lookup
- case EqualNullSafe(attrRef: AttributeReference, litNull @ Literal(null,
_)) =>
+ // Filter "colA <=> null"
+ // Translates to "colA_nullCount IS NULL OR colA_nullCount > 0" for
index lookup
+ // (this is equivalent to "colA is null")
+ case EqualNullSafe(attrRef: AttributeReference, Literal(null, _)) =>
getTargetIndexedColumnName(attrRef, indexedCols)
- .map(colName => EqualTo(genColNumNullsExpr(colName), litNull))
+ .map(colName => genColumnIsNullExpression(colName))
.orElse({
Option.empty
})
@@ -250,17 +251,19 @@ object DataSkippingUtils extends Logging {
})
// Filter "colA is null"
- // Translates to "colA_nullCount > 0" for index lookup
+ // Translates to "colA_nullCount IS NULL OR colA_nullCount > 0" for
index lookup
+ // "colA_nullCount IS NULL" means we are not certain whether the column
holds nulls or not,
+ // hence we keep the file to ensure this does not affect the query.
case IsNull(attribute: AttributeReference) =>
getTargetIndexedColumnName(attribute, indexedCols)
- .map(colName => GreaterThan(genColNumNullsExpr(colName), Literal(0)))
+ .map(colName => genColumnIsNullExpression(colName))
.orElse({
Option.empty
})
// Filter "colA is not null"
- // Translates to "colA_nullCount = null or colA_valueCount = null or
colA_nullCount < colA_valueCount" for index lookup
- // "colA_nullCount = null or colA_valueCount = null" means we are not
certain whether the column is null or not,
+ // Translates to "colA_nullCount IS NULL OR colA_valueCount IS NULL OR
colA_nullCount < colA_valueCount" for index lookup
+ // "colA_nullCount IS NULL OR colA_valueCount IS NULL" means we are not
certain whether the column is null or not,
// hence we return True to ensure this does not affect the query.
case IsNotNull(attribute: AttributeReference) =>
getTargetIndexedColumnName(attribute, indexedCols)
@@ -431,19 +434,19 @@ object DataSkippingUtils extends Logging {
// If Expression is not resolved, we can't perform the analysis
accurately, bailing
case expr if !expr.resolved => false
- // Filter "colA = null"
- // Translates to "colA_nullCount = null" for index lookup
+ // Filter "colA <=> null"
+ // Translates to "colA_nullCount IS NULL OR colA_nullCount > 0" for
index lookup
case EqualNullSafe(attrRef: AttributeReference, litNull@Literal(null,
_)) =>
getTargetIndexedColumnName(attrRef, indexedCols).isDefined
// Filter "colA is null"
- // Translates to "colA_nullCount > 0" for index lookup
+ // Translates to "colA_nullCount IS NULL OR colA_nullCount > 0" for
index lookup
case IsNull(attribute: AttributeReference) =>
getTargetIndexedColumnName(attribute, indexedCols).isDefined
// Filter "colA is not null"
- // Translates to "colA_nullCount = null or colA_valueCount = null or
colA_nullCount < colA_valueCount" for index lookup
- // "colA_nullCount = null or colA_valueCount = null" means we are not
certain whether the column is null or not,
+ // Translates to "colA_nullCount IS NULL OR colA_valueCount IS NULL OR
colA_nullCount < colA_valueCount" for index lookup
+ // "colA_nullCount IS NULL OR colA_valueCount IS NULL" means we are not
certain whether the column is null or not,
// hence we return True to ensure this does not affect the query.
case IsNotNull(attribute: AttributeReference) =>
getTargetIndexedColumnName(attribute, indexedCols).isDefined
@@ -479,6 +482,16 @@ object ColumnStatsExpressionUtils {
@inline def genColNumNullsExpr(colName: String): Expression =
sparkAdapter.getExpressionFromColumn(col(getNullCountColumnNameFor(colName)))
@inline def genColValueCountExpr: Expression =
sparkAdapter.getExpressionFromColumn(col(getValueCountColumnNameFor))
+ @inline def genColumnIsNullExpression(colName: String): Expression = {
+ val numNullsExpr = genColNumNullsExpr(colName)
+ // NOTE: Column Stats Index isn't guaranteed to hold stats for every
column of every file: for ex,
+ // a column added by schema evolution is missing from the stats of
the files written before it,
+ // and columns of types not supported by the index hold no stats at
all. In that case the
+ // null-count is null, meaning we can't tell whether the file holds
null values, and therefore
+ // such file could NOT be pruned
+ Or(IsNull(numNullsExpr), GreaterThan(numNullsExpr, Literal(0)))
+ }
+
@inline def genColumnValuesEqualToExpression(colName: String,
value: Expression,
targetExprBuilder:
Function[Expression, Expression] = Predef.identity): Expression = {
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
index 8f35e058cdd0..a86778645e13 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestDataSkippingUtils.scala
@@ -229,6 +229,35 @@ object TestDataSkippingUtils {
IndexRow("file_3", valueCount = 2, A_minValue = 2, A_maxValue = 3,
A_nullCount = 0, B_minValue = null, B_maxValue = null, B_nullCount = null)
),
Seq("file_1", "file_2")
+ ),
+ // NOTE: file_1 holds no stats for B (for ex, B was added by schema
evolution after it was written),
+ // hence it could hold nulls for B and could NOT be pruned
+ arguments(
+ "B is null",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = 0, A_maxValue = 1,
A_nullCount = 0, B_minValue = null, B_maxValue = null, B_nullCount = null),
+ IndexRow("file_2", valueCount = 2, A_minValue = 1, A_maxValue = 2,
A_nullCount = 0, B_minValue = "a", B_maxValue = "b", B_nullCount = 0),
+ IndexRow("file_3", valueCount = 2, A_minValue = 2, A_maxValue = 3,
A_nullCount = 0, B_minValue = "a", B_maxValue = "b", B_nullCount = 1)
+ ),
+ Seq("file_1", "file_3")
+ ),
+ arguments(
+ "B <=> null",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = 0, A_maxValue = 1,
A_nullCount = 0, B_minValue = null, B_maxValue = null, B_nullCount = null),
+ IndexRow("file_2", valueCount = 2, A_minValue = 1, A_maxValue = 2,
A_nullCount = 0, B_minValue = "a", B_maxValue = "b", B_nullCount = 0),
+ IndexRow("file_3", valueCount = 2, A_minValue = 2, A_maxValue = 3,
A_nullCount = 0, B_minValue = "a", B_maxValue = "b", B_nullCount = 1)
+ ),
+ Seq("file_1", "file_3")
+ ),
+ arguments(
+ "A = 1 and B is null",
+ Seq(
+ IndexRow("file_1", valueCount = 2, A_minValue = 0, A_maxValue = 1,
A_nullCount = 0, B_minValue = null, B_maxValue = null, B_nullCount = null),
+ IndexRow("file_2", valueCount = 2, A_minValue = 2, A_maxValue = 3,
A_nullCount = 0, B_minValue = null, B_maxValue = null, B_nullCount = null),
+ IndexRow("file_3", valueCount = 2, A_minValue = 1, A_maxValue = 2,
A_nullCount = 0, B_minValue = "a", B_maxValue = "b", B_nullCount = 0)
+ ),
+ Seq("file_1")
)
)
}
