cshuo commented on code in PR #20328:
URL: https://github.com/apache/hudi/pull/20328#discussion_r4367391516
##########
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:
Fixed by quoting the literal statistics column names in both the translated
expressions and the missing-statistics guards. Added mixed-coverage tests for
dotted column names, including negated predicates and an independent conjunct
that can still prune.
##########
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:
Updated the regression test to cover all four combinations of COW/MOR and
partition statistics on/off. Also added same-partition append coverage to
verify incomplete partition statistics are invalidated, and a MOR case where
the matching row comes from a log file with missing column statistics.
--
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]
