This is an automated email from the ASF dual-hosted git repository. vhs pushed a commit to branch release-1.0.2 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 859b173101e4ce4cf9ae2f2de5bce5688a671874 Author: Lokesh Jain <[email protected]> AuthorDate: Wed Apr 2 23:13:57 2025 +0530 [HUDI-9252] BaseHoodieWriteClient should only check for upgrade (#13073) (cherry picked from commit d33f95497b6ff6405bc14fc1135e7fb9d7468460) --- .../apache/hudi/client/BaseHoodieWriteClient.java | 10 +++--- .../hudi/table/upgrade/UpgradeDowngrade.java | 13 ++++++-- .../hudi/table/upgrade/TestUpgradeDowngrade.java | 39 ++++++++++++++++++++++ 3 files changed, 55 insertions(+), 7 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java index 560ffada1eb..2aed86a2440 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java @@ -957,7 +957,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient * Starts a new commit time for a write operation (insert/update/delete) with specified action. */ private void startCommitWithTime(String instantTime, String actionType, HoodieTableMetaClient metaClient) { - if (needsUpgradeOrDowngrade(metaClient)) { + if (needsUpgrade(metaClient)) { // unclear what instant to use, since upgrade does have a given instant. executeUsingTxnManager(Option.empty(), () -> tryUpgrade(metaClient, Option.empty())); } @@ -1290,7 +1290,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient ownerInstant = Option.of(metaClient.createNewInstant(HoodieInstant.State.INFLIGHT, CommitUtils.getCommitActionType(operationType, metaClient.getTableType()), instantTime.get())); } - boolean requiresInitTable = needsUpgradeOrDowngrade(metaClient) || config.isMetadataTableEnabled(); + boolean requiresInitTable = needsUpgrade(metaClient) || config.isMetadataTableEnabled(); if (!requiresInitTable) { return; } @@ -1470,7 +1470,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient UpgradeDowngrade upgradeDowngrade = new UpgradeDowngrade(metaClient, config, context, upgradeDowngradeHelper); - if (upgradeDowngrade.needsUpgradeOrDowngrade(config.getWriteVersion())) { + if (upgradeDowngrade.needsUpgrade(config.getWriteVersion())) { // Ensure no inflight commits by setting EAGER policy and explicitly cleaning all failed commits List<String> instantsToRollback = tableServiceClient.getInstantsToRollback(metaClient, HoodieFailedWritesCleaningPolicy.EAGER, instantTime); @@ -1489,9 +1489,9 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient } } - private boolean needsUpgradeOrDowngrade(HoodieTableMetaClient metaClient) { + private boolean needsUpgrade(HoodieTableMetaClient metaClient) { UpgradeDowngrade upgradeDowngrade = new UpgradeDowngrade(metaClient, config, context, upgradeDowngradeHelper); - return upgradeDowngrade.needsUpgradeOrDowngrade(config.getWriteVersion()); + return upgradeDowngrade.needsUpgrade(config.getWriteVersion()); } /** diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeDowngrade.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeDowngrade.java index 124cdad0a4c..3dafc88b733 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeDowngrade.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/UpgradeDowngrade.java @@ -69,6 +69,11 @@ public class UpgradeDowngrade { } public static boolean needsUpgradeOrDowngrade(HoodieTableMetaClient metaClient, HoodieWriteConfig config, HoodieTableVersion toWriteVersion) { + HoodieTableVersion fromTableVersion = metaClient.getTableConfig().getTableVersion(); + return needsUpgrade(metaClient, config, toWriteVersion) || toWriteVersion.versionCode() < fromTableVersion.versionCode(); + } + + public static boolean needsUpgrade(HoodieTableMetaClient metaClient, HoodieWriteConfig config, HoodieTableVersion toWriteVersion) { HoodieTableVersion fromTableVersion = metaClient.getTableConfig().getTableVersion(); // If table version is less than SIX, then we need to upgrade to SIX first before upgrading to any other version, irrespective of autoUpgrade flag if (fromTableVersion.versionCode() < HoodieTableVersion.SIX.versionCode() && toWriteVersion.versionCode() >= HoodieTableVersion.EIGHT.versionCode()) { @@ -82,14 +87,18 @@ public class UpgradeDowngrade { String.format("Please disable metadata table before upgrading from version %s to %s.", fromTableVersion, toWriteVersion)); } - // allow upgrades/downgrades otherwise. - return toWriteVersion.versionCode() != fromTableVersion.versionCode(); + // allow upgrades otherwise. + return toWriteVersion.versionCode() > fromTableVersion.versionCode(); } public boolean needsUpgradeOrDowngrade(HoodieTableVersion toWriteVersion) { return needsUpgradeOrDowngrade(metaClient, config, toWriteVersion); } + public boolean needsUpgrade(HoodieTableVersion toWriteVersion) { + return needsUpgrade(metaClient, config, toWriteVersion); + } + /** * Perform Upgrade or Downgrade steps if required and updated table version if need be. * <p> diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java index 447175f84d5..c5357f181a7 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java @@ -104,6 +104,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; /** * Unit tests {@link UpgradeDowngrade}. @@ -668,6 +670,43 @@ public class TestUpgradeDowngrade extends HoodieClientTestBase { */ } + @Test + void testNeedsUpgrade() { + HoodieTableConfig tableConfig = mock(HoodieTableConfig.class); + when(tableConfig.getTableVersion()).thenReturn(HoodieTableVersion.EIGHT); + HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class); + when(metaClient.getTableConfig()).thenReturn(tableConfig); + HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class); + when(writeConfig.autoUpgrade()).thenReturn(true); + + // assert no downgrade for table version 7 from table version 8 + boolean shouldDowngrade = new UpgradeDowngrade(metaClient, writeConfig, context, null) + .needsUpgrade(HoodieTableVersion.SEVEN); + assertFalse(shouldDowngrade); + + // assert no downgrade for table version 6 from table version 8 + shouldDowngrade = new UpgradeDowngrade(metaClient, writeConfig, context, null) + .needsUpgrade(HoodieTableVersion.SIX); + assertFalse(shouldDowngrade); + + // assert no upgrade/downgrade for table version 8 from table version 8 + shouldDowngrade = new UpgradeDowngrade(metaClient, writeConfig, context, null) + .needsUpgrade(HoodieTableVersion.EIGHT); + assertFalse(shouldDowngrade); + + // test upgrade from table version six + when(tableConfig.getTableVersion()).thenReturn(HoodieTableVersion.SIX); + // assert upgrade for table version 7 from table version 6 + boolean shouldUpgrade = new UpgradeDowngrade(metaClient, writeConfig, context, null) + .needsUpgrade(HoodieTableVersion.SEVEN); + assertTrue(shouldUpgrade); + + // assert upgrade for table version 8 from table version 6 + shouldUpgrade = new UpgradeDowngrade(metaClient, writeConfig, context, null) + .needsUpgrade(HoodieTableVersion.EIGHT); + assertTrue(shouldUpgrade); + } + private void assertMarkerFilesForDowngrade(HoodieTable table, HoodieInstant commitInstant, boolean assertExists) throws IOException { // Verify recreated marker files are as expected WriteMarkers writeMarkers = WriteMarkersFactory.get(getConfig().getMarkersType(), table, commitInstant.requestedTime());
