(hudi) branch release-1.1.2 updated: test(spark): validate cleaned files in partition stats restore test

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

danny0405 pushed a commit to branch release-1.1.2
in repository https://gitbox.apache.org/repos/asf/hudi.git


The following commit(s) were added to refs/heads/release-1.1.2 by this push:
     new 9d17c9f78e2c test(spark): validate cleaned files in partition stats restore test
9d17c9f78e2c is described below

commit 9d17c9f78e2c6ca070476a673d4700bc048c8bed
Author: danny0405 <yu...@gmail.com>
AuthorDate: Mon Oct 12 10:58:55 2026 +0800

    test(spark): validate cleaned files in partition stats restore test
    
    Savepoints can retain first-commit file slices in merge-on-read tables. Check column stats against successful deletions from completed cleans instead of assuming every first-commit file was removed. Require nonempty deletions and column stats so the check cannot pass vacuously.
---
 .../hudi/functional/TestPartitionStatsIndex.scala  | 27 ++++++++++++----------
 1 file changed, 15 insertions(+), 12 deletions(-)

diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionStatsIndex.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionStatsIndex.scala
index 802c625ceebf..b024fe94aa38 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionStatsIndex.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestPartitionStatsIndex.scala
@@ -383,9 +383,9 @@ class TestPartitionStatsIndex extends PartitionStatsIndexTestBase {
    * 1. Enable column_stats, partition_stats and record_index (files/RLI already enabled by default).
    * 2. Do two inserts and validate index initialization.
    * 3. Do a savepoint on the second commits.
-   * 4. Add three more commits to trigger clean, which cleans the files from the first commit.
+   * 4. Add three more commits to trigger clean and record the files actually deleted.
    * 5. Restore, and validate that partition_stats is deleted, but column_stats partition exists.
-   * 6. Validate that column_stats does not contain records with file names from first commit.
+   * 6. Validate that column_stats does not contain records for the deleted files.
    */
   @Test
   def testPartitionStatsWithRestore(): Unit = {
@@ -400,7 +400,6 @@ class TestPartitionStatsIndex extends PartitionStatsIndexTestBase {
       hudiOpts,
       operation = DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL,
       saveMode = SaveMode.Overwrite)
-    val firstCompletedInstant = metaClient.getActiveTimeline.getCommitsTimeline.filterCompletedInstants().lastInstant()
     // Second ingest.
     doWriteAndValidateDataAndPartitionStats(
       hudiOpts,
@@ -436,13 +435,16 @@ class TestPartitionStatsIndex extends PartitionStatsIndexTestBase {
         saveMode = SaveMode.Append)
     }
     metaClient = HoodieTableMetaClient.reload(metaClient)
-    // Clean commit should be triggered.
-    val cleanInstantOpt = metaClient.getActiveTimeline.getCleanerTimeline.lastInstant()
-    assertTrue(cleanInstantOpt.isPresent)
-    val cleanMetadataBytes = metaClient.getActiveTimeline.getInstantDetails(cleanInstantOpt.get)
-    val cleanMetadata = deserializeAvroMetadataLegacy(cleanMetadataBytes.get(), classOf[HoodieCleanMetadata])
-    // This clean operation deletes 6 files created by the first commit.
-    assertTrue(cleanMetadata.getTotalFilesDeleted > 0)
+    // A savepoint can retain first-commit file slices in MOR tables. Check the files
+    // actually deleted by all completed cleans instead of assuming their commit time.
+    val cleanTimeline = metaClient.getActiveTimeline.getCleanerTimeline.filterCompletedInstants()
+    assertFalse(cleanTimeline.empty())
+    val deletedFileNames = cleanTimeline.getInstants.asScala.flatMap { instant =>
+      val cleanMetadata = deserializeAvroMetadataLegacy(
+        cleanTimeline.getInstantDetails(instant).get(), classOf[HoodieCleanMetadata])
+      cleanMetadata.getPartitionMetadata.values().asScala.flatMap(_.getSuccessDeleteFiles.asScala)
+    }.toSet
+    assertTrue(deletedFileNames.nonEmpty, "The test must clean files before restoring")
     // Restore to savepoint
     writeClient.restoreToSavepoint(savepointTimestamp)
     // Verify restore completed
@@ -486,9 +488,10 @@ class TestPartitionStatsIndex extends PartitionStatsIndexTestBase {
     val columnStats = columnStatsIndex
       .loadColumnStatsIndexRecords(targetColumnsToIndex, shouldReadInMemory = true)
       .collectAsList()
-    // All files from first commit have been removed.
+    assertFalse(columnStats.isEmpty, "Column stats must be present after restoring and writing")
     for (stats <- columnStats.asScala) {
-      assertFalse(stats.getFileName.contains(firstCompletedInstant.get().requestedTime()))
+      assertFalse(deletedFileNames.contains(stats.getFileName),
+        s"Column stats must not retain cleaned file ${stats.getFileName}")
     }
   }