voonhous commented on code in PR #19471:
URL: https://github.com/apache/hudi/pull/19471#discussion_r3703930790
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionBucketIndexSupport.scala:
##########
@@ -173,32 +172,31 @@ class TestPartitionBucketIndexSupport extends
TestBucketIndexSupport {
def exprFilePathAnswerCheck(bucketIndexSupport: PartitionBucketIndexSupport,
exprRaw: String, expectResult: Set[String],
allFileStatus: Set[String]): Unit = {
- if (!HoodieSparkUtils.gteqSpark4_0) { // TODO (HUDI-9403)
- val resolveExpr = HoodieCatalystExpressionUtils.resolveExpr(spark,
exprRaw, structSchema)
- val optimizerPlan =
spark.sessionState.optimizer.execute(DummyExpressionHolder(Seq(resolveExpr)))
- val optimizerExpr =
optimizerPlan.asInstanceOf[DummyExpressionHolder].exprs.head
-
- // split input files into different partitions
- val partitionPath1 = DEFAULT_PARTITION_PATH(0)
- val allFileSlices1: Seq[FileSlice] = allFileStatus.slice(0,
3).map(fileName => {
- val slice = new FileSlice(partitionPath1, "00000000000000000",
FSUtils.getFileId(fileName))
- slice.setBaseFile(new HoodieBaseFile(new StoragePathInfo(new
StoragePath(fileName), 0L, false, 0, 0, 0)))
- slice
- }).toSeq
-
- val partitionPath2 = DEFAULT_PARTITION_PATH(1)
- val allFileSlices2: Seq[FileSlice] = allFileStatus.slice(3,
5).map(fileName => {
- val slice = new FileSlice(partitionPath1, "00000000000000000",
FSUtils.getFileId(fileName))
- slice.setBaseFile(new HoodieBaseFile(new StoragePathInfo(new
StoragePath(fileName), 0L, false, 0, 0, 0)))
- slice
- }).toSeq
-
- val input = Seq((Option.apply(new
BaseHoodieTableFileIndex.PartitionPath(partitionPath1, Array())),
allFileSlices1),
- (Option.apply(new
BaseHoodieTableFileIndex.PartitionPath(partitionPath2, Array())),
allFileSlices2))
- val candidate = bucketIndexSupport.computeCandidateFileNames(fileIndex,
splitConjunctivePredicates(optimizerExpr),
- Seq(), input, false)
-
- assert(candidate.get.equals(expectResult))
- }
+ val resolveExpr = HoodieCatalystExpressionUtils.resolveExpr(spark,
exprRaw, structSchema)
+ val dummyExpressionHolder = HoodieDummyExpressionHolder(Seq(resolveExpr),
resolveExpr.references.toSeq)
+ val optimizerPlan =
spark.sessionState.optimizer.execute(dummyExpressionHolder)
+ val optimizerExpr =
optimizerPlan.asInstanceOf[HoodieDummyExpressionHolder].exprs.head
+
+ // split input files into different partitions
+ val partitionPath1 = DEFAULT_PARTITION_PATH(0)
+ val allFileSlices1: Seq[FileSlice] = allFileStatus.slice(0,
3).map(fileName => {
+ val slice = new FileSlice(partitionPath1, "00000000000000000",
FSUtils.getFileId(fileName))
+ slice.setBaseFile(new HoodieBaseFile(new StoragePathInfo(new
StoragePath(fileName), 0L, false, 0, 0, 0)))
+ slice
+ }).toSeq
+
+ val partitionPath2 = DEFAULT_PARTITION_PATH(1)
+ val allFileSlices2: Seq[FileSlice] = allFileStatus.slice(3,
5).map(fileName => {
Review Comment:
The restored assertions never exercise **per-partition bucket counts** --
the one behaviour that distinguishes `PartitionBucketIndexSupport` from
`BucketIndexSupport`.
`DEFAULT_EXPRESSIONS` (line 42) is
`\d{4}\-(06\-(01|17|18)|11\-(01|10|11)),19` and `RegexRuleEngine` matches with
`Matcher.find()`, so **both** `DEFAULT_PARTITION_PATH` entries resolve to 19
buckets:
| partition | numBuckets |
| --- | --- |
| `2025-06-17` | 19 |
| `2025-06-18` | 19 |
| `2025-07-01` | 10 (default -- never reached) |
`calc` is consumed in exactly one place
(`PartitionBucketIndexSupport.scala:72`), so all 11 assertions still pass if
`computeNumBuckets` were hardcoded to `return 19`. They do discriminate 19 from
the table default of 10, so they are not vacuous -- but the per-partition
mapping itself is untested.
Two related gaps worth closing in the same pass:
1. **No fallback case.** The parent pins `bucketSet.isEmpty` 14 times via
`fallback = true`; this subclass has none. That gap is exactly why #19487
shipped unnoticed.
2. **The partition split is random per run.**
`BucketIdentifier.newBucketFileIdPrefix` -> `FSUtils.createNewFileIdPfx()` =
`UUID.randomUUID()`, so the file names differ every run and `slice(0, 3)` /
`slice(3, 5)` put different buckets in each partition each time -- over six
runs partition 2 held `5,8 / 7,8 / 4,8 / 6,7 / 4,5 / 6,8`. It is invariant
today only because both partitions share 19 buckets.
Concretely: add a third input partition whose path does **not** match the
expression (e.g. `2025-07-01`, which falls back to 10 buckets), give each
partition an explicit named file set instead of `slice`, and assert the union
across mixed bucket counts. That assertion fails if `computeNumBuckets` ever
returns the wrong count for a non-matching partition -- today nothing would.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionBucketIndexSupport.scala:
##########
@@ -173,32 +172,31 @@ class TestPartitionBucketIndexSupport extends
TestBucketIndexSupport {
def exprFilePathAnswerCheck(bucketIndexSupport: PartitionBucketIndexSupport,
exprRaw: String, expectResult: Set[String],
allFileStatus: Set[String]): Unit = {
- if (!HoodieSparkUtils.gteqSpark4_0) { // TODO (HUDI-9403)
- val resolveExpr = HoodieCatalystExpressionUtils.resolveExpr(spark,
exprRaw, structSchema)
- val optimizerPlan =
spark.sessionState.optimizer.execute(DummyExpressionHolder(Seq(resolveExpr)))
- val optimizerExpr =
optimizerPlan.asInstanceOf[DummyExpressionHolder].exprs.head
-
- // split input files into different partitions
- val partitionPath1 = DEFAULT_PARTITION_PATH(0)
- val allFileSlices1: Seq[FileSlice] = allFileStatus.slice(0,
3).map(fileName => {
- val slice = new FileSlice(partitionPath1, "00000000000000000",
FSUtils.getFileId(fileName))
- slice.setBaseFile(new HoodieBaseFile(new StoragePathInfo(new
StoragePath(fileName), 0L, false, 0, 0, 0)))
- slice
- }).toSeq
-
- val partitionPath2 = DEFAULT_PARTITION_PATH(1)
- val allFileSlices2: Seq[FileSlice] = allFileStatus.slice(3,
5).map(fileName => {
- val slice = new FileSlice(partitionPath1, "00000000000000000",
FSUtils.getFileId(fileName))
- slice.setBaseFile(new HoodieBaseFile(new StoragePathInfo(new
StoragePath(fileName), 0L, false, 0, 0, 0)))
- slice
- }).toSeq
-
- val input = Seq((Option.apply(new
BaseHoodieTableFileIndex.PartitionPath(partitionPath1, Array())),
allFileSlices1),
- (Option.apply(new
BaseHoodieTableFileIndex.PartitionPath(partitionPath2, Array())),
allFileSlices2))
- val candidate = bucketIndexSupport.computeCandidateFileNames(fileIndex,
splitConjunctivePredicates(optimizerExpr),
- Seq(), input, false)
-
- assert(candidate.get.equals(expectResult))
- }
+ val resolveExpr = HoodieCatalystExpressionUtils.resolveExpr(spark,
exprRaw, structSchema)
+ val dummyExpressionHolder = HoodieDummyExpressionHolder(Seq(resolveExpr),
resolveExpr.references.toSeq)
+ val optimizerPlan =
spark.sessionState.optimizer.execute(dummyExpressionHolder)
+ val optimizerExpr =
optimizerPlan.asInstanceOf[HoodieDummyExpressionHolder].exprs.head
Review Comment:
Nit, feel free to ignore: these four lines are now a **third**
byte-identical copy -- `TestBucketIndexSupport.scala:383-386` and `:405-408`
are the same block.
The drift risk is not hypothetical: `76bb580ae149` updated two of the three
copies and wrapped the third in the `gteqSpark4_0` guard, which is the bug this
PR is fixing.
Worth extracting into the parent so the next Spark bump has one edit site
instead of three:
```scala
protected def optimizeResolvedExpr(exprRaw: String): Expression = {
val resolveExpr = HoodieCatalystExpressionUtils.resolveExpr(spark,
exprRaw, structSchema)
val holder = HoodieDummyExpressionHolder(Seq(resolveExpr),
resolveExpr.references.toSeq)
spark.sessionState.optimizer.execute(holder).asInstanceOf[HoodieDummyExpressionHolder].exprs.head
}
```
All three call sites then collapse to `val optimizerExpr =
optimizeResolvedExpr(exprRaw)`.
--
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]
