nsivabalan commented on code in PR #20328:
URL: https://github.com/apache/hudi/pull/20328#discussion_r4364755375
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/SparkBaseIndexSupport.scala:
##########
@@ -113,14 +114,44 @@ abstract class SparkBaseIndexSupport(spark: SparkSession,
}
val validIndexedColumns = getValidIndexedColumnsFunc.apply(indexDefinition)
- val indexFilter =
queryFilters.map(translateIntoColumnStatsIndexFilterExpr(_, isExpressionIndex,
validIndexedColumns)).reduce(And)
- if (indexFilter.equals(TrueLiteral)) {
- // if there are any non indexed cols or we can't translate source expr,
we have to read all files and may not benefit from col stats lookup.
- fileNamesFromPrunedPartitions
+ // Only split top-level AND predicates; keep OR and NOT subtrees intact.
+ val indexFilters =
queryFilters.flatMap(SparkBaseIndexSupport.splitConjuncts)
+ .map(translateIntoColumnStatsIndexFilterExpr(_, isExpressionIndex,
validIndexedColumns))
+ .filterNot(_.equals(TrueLiteral))
+ if (indexFilters.isEmpty) {
+ // No translated predicate can prune files, so avoid scanning the index.
+ fileNamesFromPrunedPartitions
} else {
- // only lookup in col stats if all filters are eligible to be looked up
in col stats index in MDT
+ val projectedColumns = indexDf.columns.toSet
+ val statsToNullCount = validIndexedColumns.flatMap { column =>
+ val nullCountColumn = getNullCountColumnNameFor(column)
+ if (projectedColumns.contains(nullCountColumn)) {
+ Seq(getMinColumnNameFor(column), getMaxColumnNameFor(column),
nullCountColumn)
+ .map(_ -> nullCountColumn)
+ } else {
+ Seq.empty
+ }
+ }.toMap
+ // Guard each independent conjunct using only the statistics it reads.
Complete conjuncts
+ // can still exclude a file when another conjunct has missing statistics.
+ // For example, age > 100 AND name = 'alice' becomes:
+ // (age_maxValue > 100 OR age_nullCount IS NULL)
+ // AND (
+ // (name_minValue <= 'alice' AND name_maxValue >= 'alice')
+ // OR name_nullCount IS NULL
+ // )
+ // A file with age_maxValue = 30 and age_nullCount = 0 is still excluded
even if name statistics are missing.
+ val candidateFilter = indexFilters.map { conjunct =>
+ val referencedStats = conjunct.references.map(_.name).toSet
+ val nullCountColumns = referencedStats.flatMap(statsToNullCount.get)
+ // Apply the missing-statistics guard outside the whole translated
conjunct, including NOT.
+ // Replacing a missing-statistics leaf with true before negation would
incorrectly prune files.
+
nullCountColumns.foldLeft(sparkAdapter.createColumnFromExpression(conjunct)) {
(filter, column) =>
+ filter.or(indexDf.col(column).isNull)
Review Comment:
Small question about how this column name resolves. The translated conjunct
refers to stats columns through `functions.col(...)` (`genColNumNullsExpr` and
friends in `DataSkippingUtils`), but the guard uses `indexDf.col(column)`
directly. For nested indexed fields (e.g. `c9.c9_1_car_brand` in
`TestColumnStatsIndex`), the transposed column is literally named
`c9.c9_1_car_brand_nullCount`, and `Dataset.col` splits on dots unless the name
is backtick-quoted. Using `DataSkippingUtils.genColNumNullsExpr(column)` (or
quoting the name) would make the guard resolve exactly like the expression it
protects. Could you add a test case with a mixed-coverage predicate on a nested
indexed column?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/feature/TestDataSkippingQuery.scala:
##########
@@ -128,6 +135,66 @@ class TestDataSkippingQuery extends HoodieSparkSqlTestBase
{
}
}
+ test("Data skipping retains files with missing column statistics after index
expansion") {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ withSQLConf(
+ "hoodie.metadata.enable" -> "true",
+ "hoodie.metadata.index.column.stats.enable" -> "true",
+ "hoodie.metadata.index.partition.stats.enable" -> "false",
Review Comment:
This setting isolates the file-level path, which is what we want for the
reproduction. Could you parameterize the test over partition stats on/off once
`PartitionStatsIndexSupport` gets the same fix (see review summary)? On is the
default, so that's the configuration most users run, and right now it isn't
covered. A COW/MOR parameter would also be cheap to add here.
--
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]
