intial address comments of siva

Author: rahil-c

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeDowngrade.java

@@ -151,8 +151,7 @@ public void run(HoodieTableVersion toVersion, String instantTime) {
     boolean isUpgrade = fromVersion.versionCode() < toVersion.versionCode();
     if (isUpgrade && !config.autoUpgrade()) {
       // if we are attempting to upgrade and auto-upgrade is disabled
-      // we set the write config table version to bounded by the current hudi table version
-      // and then exit out the upgrade process
+      // we exit out the upgrade process
       LOG.warn("AUTO_UPGRADE_VERSION was explicitly disabled, skipping table version upgrade process");
       return;
     }

hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java

@@ -92,6 +92,11 @@ public void testUpgradeOrDowngrade(HoodieTableVersion fromVersion, HoodieTableVe
 
     Dataset<Row> originalData = readTableData(originalMetaClient, "before " + operation);
 
+    // Confirm that there are log files before rollback and compaction operations
+    if (isRollbackAndCompactTransition(fromVersion, toVersion)) {
+      validateLogFilesCount(originalMetaClient, operation, true);
+    }
+    
     new UpgradeDowngrade(originalMetaClient, config, context(), SparkUpgradeDowngradeHelper.getInstance())
         .run(toVersion, null);
 
@@ -103,13 +108,15 @@ public void testUpgradeOrDowngrade(HoodieTableVersion fromVersion, HoodieTableVe
     assertTableVersionOnDataAndMetadataTable(resultMetaClient, toVersion);
     validateVersionSpecificProperties(resultMetaClient, fromVersion, toVersion);
     validateDataConsistency(originalData, resultMetaClient, "after " + operation);
-    
-    // Validate pending commits based on whether this transition performs rollback operations
+
+    // Validate pending commits based on whether this transition performs rollback and compaction operations
     int finalPendingCommits = resultMetaClient.getCommitsTimeline().filterPendingExcludingCompaction().countInstants();
-    if (isRollbackTransition(fromVersion, toVersion)) {
+    if (isRollbackAndCompactTransition(fromVersion, toVersion)) {
       // Handlers that call rollbackFailedWritesAndCompact() clear all pending commits
       assertEquals(0, finalPendingCommits,
           "Pending commits should be cleared to 0 after " + operation);
+      // Validate no log files remain after rollback and compaction
+      validateLogFilesCount(resultMetaClient, operation, false);
     } else {
       // Other handlers may clean up some pending commits but don't necessarily clear all
       assertTrue(finalPendingCommits <= initialPendingCommits,
@@ -187,9 +194,13 @@ public void testMetadataTableUpgradeDowngradeFailure(HoodieTableVersion fromVers
         () -> new UpgradeDowngrade(originalMetaClient, cfg, context(), SparkUpgradeDowngradeHelper.getInstance())
             .run(toVersion, null)
     );
-
-    LOG.info("Successfully caught expected exception during {} from {} to {}: {}", 
-        operation, fromVersion, toVersion, exception.getMessage());
+    
+    // Verify the specific exception message for metadata table failures
+    String expectedMessage = "Upgrade/downgrade for the Hudi metadata table failed. "
+        + "Please try again. If the failure repeats for metadata table, it is recommended to disable "
+        + "the metadata table so that the upgrade and downgrade can continue for the data table.";
+    assertTrue(exception.getMessage().contains(expectedMessage),
+        "Exception message should contain metadata table failure message");
   }
 
   /**
@@ -294,16 +305,17 @@ private static Stream<Arguments> upgradeDowngradeVersionPairs() {
   }
 
   /**
-   * Version pairs that should be tested for metadata upgrade/downgrade failures.
-   * Excludes pairs that trigger rollback operations which delete metadata tables.
+   * Version pairs for testing metadata failure when trying to upgrade/downgrade. Note these version pairs
+   * are ones that do invoke rollbackFailedWritesAndCompact() which this method causes the metadata table to be disabled
    */
   private static Stream<Arguments> metadataTableCorruptionTestVersionPairs() {
     return Stream.of(
-        Arguments.of(HoodieTableVersion.FOUR, HoodieTableVersion.FIVE),   // V4 -> V5
-        Arguments.of(HoodieTableVersion.FIVE, HoodieTableVersion.FOUR)    // V5 -> V4
-    // Note: Other pairs like V5->V6, V6->V8 are excluded because they may trigger
-    // rollback operations in rollbackFailedWritesAndCompact() that disable metadata
-    // tables, causing automatic deletion before testing can occur
+        // Non-rollback upgrade pairs
+        Arguments.of(HoodieTableVersion.FOUR, HoodieTableVersion.FIVE),   // V4 -> V5 (works)
+        Arguments.of(HoodieTableVersion.FIVE, HoodieTableVersion.SIX),    // V5 -> V6 (works)
+
+        // Non-rollback downgrade pairs  
+        Arguments.of(HoodieTableVersion.FIVE, HoodieTableVersion.FOUR)    // V5 -> V4 (works)
     );
   }
 
@@ -427,12 +439,64 @@ private boolean isMetadataTablePresent(HoodieTableMetaClient metaClient) throws
     StoragePath metadataTablePath = HoodieTableMetadata.getMetadataTableBasePath(metaClient.getBasePath());
     return metaClient.getStorage().exists(metadataTablePath);
   }
-  
+
+  /**
+   * Validate log files count based on expected scenario.
+   * This ensures proper behavior before and after rollback and compaction operations.
+   */
+  private void validateLogFilesCount(HoodieTableMetaClient metaClient, String operation, boolean expectLogFiles) {
+    String validationPhase = expectLogFiles ? "before" : "after";
+    LOG.info("Validating log files {} rollback and compaction during {}", validationPhase, operation);
+    
+    // Get the latest completed commit to ensure we're looking at a consistent state
+    org.apache.hudi.common.table.timeline.HoodieTimeline completedTimeline = 
+        metaClient.getCommitsTimeline().filterCompletedInstants();
+    String latestCommit = completedTimeline.lastInstant()
+        .map(instant -> instant.requestedTime())
+        .orElse(null);
+    
+    // Get file system view to check for log files using the latest commit state
+    try (org.apache.hudi.common.table.view.HoodieTableFileSystemView fsView = 
+        org.apache.hudi.common.table.view.HoodieTableFileSystemView.fileListingBasedFileSystemView(
+            context(), metaClient, completedTimeline)) {
+    
+      // Get all partition paths using FSUtils
+      List<String> partitionPaths = org.apache.hudi.common.fs.FSUtils.getAllPartitionPaths(
+          context(), metaClient, false);
+      
+      int totalLogFiles = 0;
+      
+      for (String partitionPath : partitionPaths) {
+        // Get latest file slices for this partition
+        Stream<org.apache.hudi.common.model.FileSlice> fileSlicesStream = latestCommit != null
+            ? fsView.getLatestFileSlicesBeforeOrOn(partitionPath, latestCommit, false)
+            : fsView.getLatestFileSlices(partitionPath);
+        
+        for (org.apache.hudi.common.model.FileSlice fileSlice : fileSlicesStream.collect(Collectors.toList())) {
+          int logFileCount = (int) fileSlice.getLogFiles().count();
+          totalLogFiles += logFileCount;
+        }
+      }
+      
+      if (expectLogFiles) {
+        assertTrue(totalLogFiles > 0, 
+            "Expected log files but found none during " + operation);
+      } else {
+        assertEquals(0, totalLogFiles, 
+            "No log files should remain after rollback and compaction during " + operation);
+      }
+      LOG.info("Log file validation passed: {} log files found (expected: {})", 
+          totalLogFiles, expectLogFiles ? ">0" : "0");
+    } catch (Exception e) {
+      throw new RuntimeException("Failed to validate log files during " + operation, e);
+    }
+  }
+
   /**
    * Determine if a version transition performs rollback operations that clear all pending commits.
    * These handlers call rollbackFailedWritesAndCompact() which clears pending commits to 0.
    */
-  private boolean isRollbackTransition(HoodieTableVersion fromVersion, HoodieTableVersion toVersion) {
+  private boolean isRollbackAndCompactTransition(HoodieTableVersion fromVersion, HoodieTableVersion toVersion) {
     // Upgrade handlers that perform rollbacks
     if (fromVersion == HoodieTableVersion.SEVEN && toVersion == HoodieTableVersion.EIGHT) {
       return true; // SevenToEightUpgradeHandler