This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 4853b57be208 fix: harden async indexing and improve coverage (#19537)
4853b57be208 is described below
commit 4853b57be208c47eaa0a976916e1130f77cb48ac
Author: Danny Chan <[email protected]>
AuthorDate: Mon Aug 10 16:47:56 2026 +0800
fix: harden async indexing and improve coverage (#19537)
* test: improve upgrade and async index coverage
* fix: address async index review feedback
---
.../action/index/AbstractIndexingCatchupTask.java | 6 +
.../table/action/index/IndexingCatchupTask.java | 5 +
.../table/action/index/RunIndexActionExecutor.java | 16 +-
.../bucket/TestConsistentBucketIndexUtils.java | 229 +++++++++++++
.../index/bucket/TestHoodieSimpleBucketIndex.java | 192 +++++++++++
.../action/index/TestIndexActionExecutors.java | 349 ++++++++++++++++++++
.../action/index/TestIndexingCatchupTask.java | 123 +++++++
.../TestLegacyUpgradeDowngradeHandlers.java | 354 +++++++++++++++++++++
.../upgrade/TestUpgradeDowngradeOrchestration.java | 236 ++++++++++++++
.../hudi/table/upgrade/TestUpgradeDowngrade.java | 56 ++++
10 files changed, 1561 insertions(+), 5 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/AbstractIndexingCatchupTask.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/AbstractIndexingCatchupTask.java
index 2ab39cd30534..fe1bd86e92f2 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/AbstractIndexingCatchupTask.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/AbstractIndexingCatchupTask.java
@@ -129,6 +129,7 @@ public abstract class AbstractIndexingCatchupTask
implements IndexingCatchupTask
default:
throw new IllegalStateException("Unexpected value: " +
instant.getAction());
}
+ currentCaughtupInstant = instantTime;
} catch (IOException e) {
throw new HoodieIndexException(String.format("Could not update
metadata partition for instant: %s", instant), e);
} finally {
@@ -138,6 +139,11 @@ public abstract class AbstractIndexingCatchupTask
implements IndexingCatchupTask
}
}
+ @Override
+ public String getCurrentCaughtupInstant() {
+ return currentCaughtupInstant;
+ }
+
/**
* Updates metadata table for the instant. This is only called for actions
that do actual writes,
* i.e. for commit/deltacommit/compaction/replacecommit and not for
clean/restore/rollback actions.
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/IndexingCatchupTask.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/IndexingCatchupTask.java
index 5d07175c3a93..e5e63c430ceb 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/IndexingCatchupTask.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/IndexingCatchupTask.java
@@ -31,6 +31,11 @@ import java.io.IOException;
*/
public interface IndexingCatchupTask extends Runnable {
+ /**
+ * Returns the latest instant successfully caught up by this task.
+ */
+ String getCurrentCaughtupInstant();
+
/**
* Update the index for the write action.
*
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/RunIndexActionExecutor.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/RunIndexActionExecutor.java
index c0a29695bf45..381fab4f649e 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/RunIndexActionExecutor.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/index/RunIndexActionExecutor.java
@@ -180,9 +180,13 @@ public class RunIndexActionExecutor<T, I, K, O> extends
BaseActionExecutor<T, I,
String indexUptoInstant = fileIndexPartitionInfo.getIndexUptoInstant();
// save index commit metadata and update table config
// instantiation of metadata writer will automatically instantiate the
partitions.
- table.getIndexingMetadataWriter(instantTime)
+ try (HoodieTableMetadataWriter metadataWriter =
table.getIndexingMetadataWriter(instantTime)
.orElseThrow(() -> new HoodieIndexException(String.format(
- "Could not get metadata writer to run index action for
instant: %s", instantTime)));
+ "Could not get metadata writer to run index action for
instant: %s", instantTime)))) {
+ // Initialization is performed when the writer is instantiated.
+ } catch (Exception e) {
+ throw new HoodieMetadataException("Failed to initialize metadata
table", e);
+ }
finalIndexPartitionInfos = Stream.of(fileIndexPartitionInfo)
.map(info -> new HoodieIndexPartitionInfo(
info.getVersion(),
@@ -281,13 +285,15 @@ public class RunIndexActionExecutor<T, I, K, O> extends
BaseActionExecutor<T, I,
HoodieHeartbeatClient heartbeatClient = new
HoodieHeartbeatClient(table.getStorage(),
table.getMetaClient().getBasePath().toString(),
table.getConfig().getHoodieClientHeartbeatIntervalInMs(),
table.getConfig().getHoodieClientHeartbeatTolerableMisses());
ExecutorService executorService =
Executors.newFixedThreadPool(MAX_CONCURRENT_INDEXING);
- Future<?> indexingCatchupTaskFuture = executorService.submit(
- IndexingCatchupTaskFactory.createCatchupTask(indexPartitionInfos,
metadataWriter, instantsToIndex, metadataCompletedTimestamps,
- table, metadataMetaClient, currentCaughtupInstant, txnManager,
context, heartbeatClient));
+ IndexingCatchupTask indexingCatchupTask =
IndexingCatchupTaskFactory.createCatchupTask(
+ indexPartitionInfos, metadataWriter, instantsToIndex,
metadataCompletedTimestamps,
+ table, metadataMetaClient, currentCaughtupInstant, txnManager,
context, heartbeatClient);
+ Future<?> indexingCatchupTaskFuture =
executorService.submit(indexingCatchupTask);
try {
log.info("Starting index catchup task");
HoodieTimer timer = HoodieTimer.start();
indexingCatchupTaskFuture.get(config.getIndexingCheckTimeoutSeconds(),
TimeUnit.SECONDS);
+ currentCaughtupInstant = indexingCatchupTask.getCurrentCaughtupInstant();
metrics.ifPresent(m ->
m.updateMetrics(HoodieMetadataMetrics.ASYNC_INDEXER_CATCHUP_TIME,
timer.endTimer()));
} catch (Exception e) {
indexingCatchupTaskFuture.cancel(true);
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestConsistentBucketIndexUtils.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestConsistentBucketIndexUtils.java
new file mode 100644
index 000000000000..38f636de70a6
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestConsistentBucketIndexUtils.java
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.index.bucket;
+
+import org.apache.hudi.common.model.ConsistentHashingNode;
+import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.view.TableFileSystemView.BaseFileOnlyView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieIndexException;
+import org.apache.hudi.io.util.FileIOUtils;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.table.HoodieTable;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.io.ByteArrayInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static
org.apache.hudi.common.model.HoodieConsistentHashingMetadata.HASHING_METADATA_COMMIT_FILE_SUFFIX;
+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.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class TestConsistentBucketIndexUtils {
+
+ private HoodieTable table;
+ private HoodieTableMetaClient metaClient;
+ private HoodieStorage storage;
+ private StoragePath hashingMetadataPath;
+
+ @BeforeEach
+ void setUp() {
+ table = mock(HoodieTable.class);
+ metaClient = mock(HoodieTableMetaClient.class);
+ storage = mock(HoodieStorage.class);
+ hashingMetadataPath = new StoragePath("/table/.hoodie/.hashing_metadata");
+ when(table.getMetaClient()).thenReturn(metaClient);
+ when(table.getStorage()).thenReturn(storage);
+ when(metaClient.getStorage()).thenReturn(storage);
+
when(metaClient.getHashingMetadataPath()).thenReturn(hashingMetadataPath.toString());
+ }
+
+ @Test
+ void testLoadMetadataReturnsEmptyForMissingPartitionAndWrapsIoFailure()
throws Exception {
+ StoragePath partitionPath = new StoragePath(hashingMetadataPath, "p");
+ doThrow(new FileNotFoundException("missing"))
+ .doThrow(new IOException("storage failure"))
+ .when(storage).listDirectEntries(partitionPath);
+ assertFalse(ConsistentBucketIndexUtils.loadMetadata(table,
"p").isPresent());
+
+ assertThrows(HoodieIndexException.class, () ->
ConsistentBucketIndexUtils.loadMetadata(table, "p"));
+ }
+
+ @Test
+ void testLoadMetadataReadsInitialAndLatestCommittedFiles() throws Exception {
+ HoodieConsistentHashingMetadata initial = new
HoodieConsistentHashingMetadata("p", 4);
+ StoragePath partitionPath = new StoragePath(hashingMetadataPath, "p");
+ StoragePathInfo initialFile = file(partitionPath, initial.getFilename());
+
when(storage.listDirectEntries(partitionPath)).thenReturn(Collections.singletonList(initialFile));
+ when(storage.open(initialFile.getPath())).thenReturn(new
ByteArrayInputStream(initial.toBytes()));
+
+ Option<HoodieConsistentHashingMetadata> loaded =
ConsistentBucketIndexUtils.loadMetadata(table, "p");
+ assertTrue(loaded.isPresent());
+ assertEquals(initial.getFilename(), loaded.get().getFilename());
+ assertEquals(4, loaded.get().getNumBuckets());
+
+ HoodieConsistentHashingMetadata updated = new
HoodieConsistentHashingMetadata(
+ (short) 0, "p", "002", 0, 0, Collections.emptyList());
+ StoragePathInfo updatedFile = file(partitionPath, updated.getFilename());
+ StoragePathInfo commitMarker = file(partitionPath, "002" +
HASHING_METADATA_COMMIT_FILE_SUFFIX);
+
when(storage.listDirectEntries(partitionPath)).thenReturn(Arrays.asList(initialFile,
updatedFile, commitMarker));
+ when(storage.open(updatedFile.getPath())).thenReturn(new
ByteArrayInputStream(updated.toBytes()));
+
+ loaded = ConsistentBucketIndexUtils.loadMetadata(table, "p");
+ assertTrue(loaded.isPresent());
+ assertEquals("002", loaded.get().getInstant());
+ }
+
+ @Test
+ void testSaveMetadataHandlesExistingConcurrentAndFailedWrites() throws
Exception {
+ HoodieConsistentHashingMetadata metadata = new
HoodieConsistentHashingMetadata("p", 4);
+ StoragePath fullPath = new StoragePath(new
StoragePath(hashingMetadataPath, "p"), metadata.getFilename());
+ when(storage.exists(fullPath)).thenReturn(true);
+ assertTrue(ConsistentBucketIndexUtils.saveMetadata(table, metadata));
+
+ when(storage.exists(fullPath)).thenReturn(false);
+ assertTrue(ConsistentBucketIndexUtils.saveMetadata(table, metadata));
+ verify(storage).createImmutableFileInPath(eq(fullPath), any(Option.class),
eq(true));
+
+ doThrow(new IOException("concurrent
create")).doReturn(true).when(storage).exists(fullPath);
+ assertTrue(ConsistentBucketIndexUtils.saveMetadata(table, metadata));
+
+ doThrow(new IOException("failed
create")).doReturn(false).when(storage).exists(fullPath);
+ assertFalse(ConsistentBucketIndexUtils.saveMetadata(table, metadata));
+ }
+
+ @Test
+ void testLoadOrCreatePersistsNewMetadata() throws Exception {
+ StoragePath partitionPath = new StoragePath(hashingMetadataPath, "p");
+ when(storage.listDirectEntries(partitionPath)).thenThrow(new
FileNotFoundException("missing"));
+ when(storage.exists(any(StoragePath.class))).thenReturn(false);
+
+ HoodieConsistentHashingMetadata metadata =
ConsistentBucketIndexUtils.loadOrCreateMetadata(table, "p", 6);
+
+ assertEquals("p", metadata.getPartitionPath());
+ assertEquals(6, metadata.getNumBuckets());
+ verify(storage).createImmutableFileInPath(any(StoragePath.class),
any(Option.class), eq(true));
+ }
+
+ @Test
+ void testLoadMetadataReturnsEmptyWhenChosenFileDisappears() throws Exception
{
+ HoodieConsistentHashingMetadata initial = new
HoodieConsistentHashingMetadata("p", 4);
+ StoragePath partitionPath = new StoragePath(hashingMetadataPath, "p");
+ StoragePathInfo initialFile = file(partitionPath, initial.getFilename());
+
when(storage.listDirectEntries(partitionPath)).thenReturn(Collections.singletonList(initialFile));
+ when(storage.open(initialFile.getPath())).thenThrow(new
FileNotFoundException("raced with cleaner"));
+
+ assertFalse(ConsistentBucketIndexUtils.loadMetadata(table,
"p").isPresent());
+ }
+
+ @Test
+ void testLoadMetadataRepairsCommitMarkerForCompletedRehash() throws
Exception {
+ HoodieConsistentHashingMetadata updated = new
HoodieConsistentHashingMetadata(
+ (short) 0, "p", "002", 0, 0, Collections.emptyList());
+ StoragePath partitionPath = new StoragePath(hashingMetadataPath, "p");
+ StoragePathInfo updatedFile = file(partitionPath, updated.getFilename());
+ StoragePath markerPath = new StoragePath(partitionPath, "002" +
HASHING_METADATA_COMMIT_FILE_SUFFIX);
+
when(storage.listDirectEntries(partitionPath)).thenReturn(Collections.singletonList(updatedFile));
+ when(storage.open(updatedFile.getPath())).thenReturn(new
ByteArrayInputStream(updated.toBytes()));
+ when(storage.exists(markerPath)).thenReturn(false);
+ HoodieActiveTimeline activeTimeline = mock(HoodieActiveTimeline.class);
+ HoodieTimeline completedTimeline = mock(HoodieTimeline.class);
+ when(metaClient.getActiveTimeline()).thenReturn(activeTimeline);
+
when(activeTimeline.getCommitAndReplaceTimeline()).thenReturn(completedTimeline);
+
when(completedTimeline.filterCompletedInstants()).thenReturn(completedTimeline);
+ when(completedTimeline.containsInstant("002")).thenReturn(true);
+
+ try (MockedStatic<FileIOUtils> fileIo = mockStatic(FileIOUtils.class,
Mockito.CALLS_REAL_METHODS)) {
+ Option<HoodieConsistentHashingMetadata> loaded =
ConsistentBucketIndexUtils.loadMetadata(table, "p");
+
+ assertTrue(loaded.isPresent());
+ assertEquals("002", loaded.get().getInstant());
+ fileIo.verify(() -> FileIOUtils.createFileInPath(storage, markerPath,
Option.empty()));
+ }
+ }
+
+ @Test
+ void testLoadOrCreateReloadsMetadataAfterConcurrentCreate() throws Exception
{
+ HoodieConsistentHashingMetadata concurrent = new
HoodieConsistentHashingMetadata("p", 3);
+ StoragePath partitionPath = new StoragePath(hashingMetadataPath, "p");
+ StoragePathInfo concurrentFile = file(partitionPath,
concurrent.getFilename());
+ when(storage.listDirectEntries(partitionPath))
+ .thenThrow(new FileNotFoundException("not created yet"))
+ .thenReturn(Collections.singletonList(concurrentFile));
+ when(storage.open(concurrentFile.getPath())).thenReturn(new
ByteArrayInputStream(concurrent.toBytes()));
+ when(storage.exists(any(StoragePath.class)))
+ .thenThrow(new IOException("lost create race"))
+ .thenReturn(false);
+
+ HoodieConsistentHashingMetadata loaded =
ConsistentBucketIndexUtils.loadOrCreateMetadata(table, "p", 9);
+
+ assertEquals(3, loaded.getNumBuckets());
+ }
+
+ @Test
+ void testLoadMetadataRejectsUncommittedRehashWithoutMatchingBaseFile()
throws Exception {
+ HoodieConsistentHashingMetadata updated = new
HoodieConsistentHashingMetadata(
+ (short) 0, "p", "002", 0, 0,
+ Collections.singletonList(new ConsistentHashingNode(100,
"new-file-group")));
+ StoragePath partitionPath = new StoragePath(hashingMetadataPath, "p");
+ StoragePathInfo updatedFile = file(partitionPath, updated.getFilename());
+
when(storage.listDirectEntries(partitionPath)).thenReturn(Collections.singletonList(updatedFile));
+ when(storage.open(updatedFile.getPath())).thenReturn(new
ByteArrayInputStream(updated.toBytes()));
+ HoodieActiveTimeline activeTimeline = mock(HoodieActiveTimeline.class);
+ HoodieTimeline completedTimeline = mock(HoodieTimeline.class);
+ HoodieTimeline pendingTimeline = mock(HoodieTimeline.class);
+ when(metaClient.getActiveTimeline()).thenReturn(activeTimeline);
+
when(activeTimeline.getCommitAndReplaceTimeline()).thenReturn(completedTimeline);
+
when(completedTimeline.filterCompletedInstants()).thenReturn(completedTimeline);
+ when(completedTimeline.containsInstant("002")).thenReturn(false);
+ when(table.getPendingCommitsTimeline()).thenReturn(pendingTimeline);
+ when(pendingTimeline.containsInstant("002")).thenReturn(false);
+ BaseFileOnlyView baseFileView = mock(BaseFileOnlyView.class);
+ when(table.getBaseFileOnlyView()).thenReturn(baseFileView);
+ when(baseFileView.getLatestBaseFiles("p")).thenAnswer(ignored ->
java.util.stream.Stream.empty());
+
+ assertFalse(ConsistentBucketIndexUtils.loadMetadata(table,
"p").isPresent());
+ }
+
+ private static StoragePathInfo file(StoragePath parent, String name) {
+ return new StoragePathInfo(new StoragePath(parent, name), 1L, false,
(short) 1, 1L, 1L);
+ }
+}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestHoodieSimpleBucketIndex.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestHoodieSimpleBucketIndex.java
new file mode 100644
index 000000000000..9995d912b9ae
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/index/bucket/TestHoodieSimpleBucketIndex.java
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.index.bucket;
+
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordLocation;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.view.TableFileSystemView;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.index.HoodieIndexUtils;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.table.HoodieTable;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Stream;
+
+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.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+class TestHoodieSimpleBucketIndex {
+
+ @Test
+ void testPendingDataFileDetectionHandlesBucketAndNonBucketFiles() {
+ HoodieSimpleBucketIndex index = new HoodieSimpleBucketIndex(config(8));
+ List<StoragePathInfo> files = Arrays.asList(
+ file("/table/p/00000003-file_002.parquet", true),
+ file("/table/p/not-a-bucket-file", true),
+ file("/table/p/00000003-file_001.parquet", false));
+
+ assertTrue(index.hasPendingDataFilesForInstant(files, "002", 3));
+ assertFalse(index.hasPendingDataFilesForInstant(files, "999", 3));
+ assertFalse(index.hasPendingDataFilesForInstant(files, "002", 4));
+ assertFalse(index.canIndexLogFiles());
+ }
+
+ @Test
+ void testFindConflictInstantsUsesInflightSlicesAndToleratesListingFailure()
throws Exception {
+ HoodieSimpleBucketIndex index = new HoodieSimpleBucketIndex(config(8));
+ HoodieTable table = mock(HoodieTable.class);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieStorage storage = mock(HoodieStorage.class);
+ TableFileSystemView.SliceView sliceView =
mock(TableFileSystemView.SliceView.class);
+ when(table.getMetaClient()).thenReturn(metaClient);
+ when(metaClient.getBasePath()).thenReturn(new StoragePath("/table"));
+ when(metaClient.getStorage()).thenReturn(storage);
+ when(table.getSliceView()).thenReturn(sliceView);
+ when(storage.listFiles(new
StoragePath("/table/p"))).thenReturn(Collections.singletonList(
+ file("/table/p/00000003-file_002.parquet", true)));
+
when(sliceView.getLatestFileSlicesIncludingInflight("p")).thenAnswer(ignored ->
Stream.of(
+ new FileSlice("p", "002", "00000003-file"),
+ new FileSlice("p", "003", "00000003-file")));
+
+ assertEquals(Collections.singletonList("002"),
+ index.findConflictInstantsInPartition(table, "p", 3, Set.of("002")));
+
+ when(storage.listFiles(any(StoragePath.class))).thenThrow(new
IOException("listing failure"));
+ assertEquals(Collections.emptyList(),
+ index.findConflictInstantsInPartition(table, "p", 3, Set.of("002")));
+ }
+
+ @Test
+ void testLoadBucketMappingAndDuplicateDetection() throws Exception {
+ HoodieSimpleBucketIndex index = new HoodieSimpleBucketIndex(config(8));
+ HoodieTable table = mock(HoodieTable.class);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline timeline = mock(HoodieActiveTimeline.class);
+ HoodieStorage storage = mock(HoodieStorage.class);
+ TableFileSystemView.SliceView sliceView =
mock(TableFileSystemView.SliceView.class);
+ when(table.getMetaClient()).thenReturn(metaClient);
+ when(metaClient.reloadActiveTimeline()).thenReturn(timeline);
+ when(timeline.filterInflights()).thenReturn(timeline);
+ when(timeline.getInstantsAsStream()).thenAnswer(ignored -> Stream.empty());
+ when(metaClient.getBasePath()).thenReturn(new StoragePath("/table"));
+ when(metaClient.getStorage()).thenReturn(storage);
+
when(storage.listFiles(any(StoragePath.class))).thenReturn(Collections.emptyList());
+ when(table.getSliceView()).thenReturn(sliceView);
+
when(sliceView.getLatestFileSlicesIncludingInflight("p")).thenAnswer(ignored ->
Stream.empty());
+ FileSlice first = new FileSlice("p", "001", "00000003-file-a");
+ FileSlice secondBucket = new FileSlice("p", "001", "00000004-file-b");
+
+ try (MockedStatic<HoodieIndexUtils> indexUtils =
mockStatic(HoodieIndexUtils.class)) {
+ indexUtils.when(() ->
HoodieIndexUtils.getLatestFileSlicesForPartition("p", table))
+ .thenReturn(Arrays.asList(first, secondBucket));
+ Map<Integer, HoodieRecordLocation> mapping =
index.loadBucketIdToFileIdMappingForPartition(table, "p");
+ assertEquals(Set.of(3, 4), mapping.keySet());
+ assertEquals("00000003-file-a", mapping.get(3).getFileId());
+
+ indexUtils.when(() ->
HoodieIndexUtils.getLatestFileSlicesForPartition("p", table))
+ .thenReturn(Arrays.asList(first, new FileSlice("p", "002",
"00000003-file-conflict")));
+ assertThrows(HoodieIOException.class,
+ () -> index.loadBucketIdToFileIdMappingForPartition(table, "p"));
+ }
+ }
+
+ @Test
+ void testIndexLocationFunctionAssignsAndLooksUpBucket() {
+ HoodieWriteConfig config = config(8);
+ HoodieKey key = new HoodieKey("record-key", "p");
+ int bucket = BucketIdentifier.getBucketId(key.getRecordKey(),
"_hoodie_record_key", 8);
+ HoodieRecordLocation expected = new HoodieRecordLocation("001",
BucketIdentifier.bucketIdStr(bucket) + "-file");
+ TestableSimpleBucketIndex index = new TestableSimpleBucketIndex(config,
Collections.singletonMap(bucket, expected));
+ HoodieTable table = mock(HoodieTable.class);
+ when(table.getConfig()).thenReturn(config);
+ HoodieRecord record = mock(HoodieRecord.class);
+ when(record.getKey()).thenReturn(key);
+ when(record.getPartitionPath()).thenReturn("p");
+
+ Function<HoodieRecord, Option<HoodieRecordLocation>> locationFunction =
index.locationFunction(table, "p");
+ assertEquals(expected, locationFunction.apply(record).get());
+
+ HoodieKey missingKey = new HoodieKey("different-key", "p");
+ while (BucketIdentifier.getBucketId(missingKey.getRecordKey(),
"_hoodie_record_key", 8) == bucket) {
+ missingKey = new HoodieKey(missingKey.getRecordKey() + "x", "p");
+ }
+ when(record.getKey()).thenReturn(missingKey);
+ assertFalse(locationFunction.apply(record).isPresent());
+ assertEquals(BucketIdentifier.getBucketId(missingKey.getRecordKey(),
"_hoodie_record_key", 8),
+ index.getBucketID(missingKey, 8));
+ }
+
+ private static HoodieWriteConfig config(int buckets) {
+ return HoodieWriteConfig.newBuilder()
+ .withPath("/table")
+ .withIndexConfig(HoodieIndexConfig.newBuilder()
+ .withBucketNum(String.valueOf(buckets))
+ .withIndexKeyField("_hoodie_record_key")
+ .build())
+ .build();
+ }
+
+ private static StoragePathInfo file(String path, boolean isFile) {
+ return new StoragePathInfo(new StoragePath(path), 1L, !isFile, (short) 1,
1L, 1L);
+ }
+
+ private static class TestableSimpleBucketIndex extends
HoodieSimpleBucketIndex {
+ private final Map<Integer, HoodieRecordLocation> mapping;
+
+ TestableSimpleBucketIndex(HoodieWriteConfig config, Map<Integer,
HoodieRecordLocation> mapping) {
+ super(config);
+ this.mapping = mapping;
+ }
+
+ @Override
+ public Map<Integer, HoodieRecordLocation>
loadBucketIdToFileIdMappingForPartition(HoodieTable table, String partition) {
+ return mapping;
+ }
+
+ Function<HoodieRecord, Option<HoodieRecordLocation>>
locationFunction(HoodieTable table, String partition) {
+ return getIndexLocationFunctionForPartition(table, partition);
+ }
+ }
+}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/index/TestIndexActionExecutors.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/index/TestIndexActionExecutors.java
new file mode 100644
index 000000000000..81a7392b1cc5
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/index/TestIndexActionExecutors.java
@@ -0,0 +1,349 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.table.action.index;
+
+import org.apache.hudi.avro.model.HoodieIndexCommitMetadata;
+import org.apache.hudi.avro.model.HoodieIndexPartitionInfo;
+import org.apache.hudi.avro.model.HoodieIndexPlan;
+import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient;
+import org.apache.hudi.client.transaction.TransactionManager;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.WriteConcurrencyMode;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieArchivedTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieLockConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.core.transaction.lock.InProcessLockProvider;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieIndexException;
+import org.apache.hudi.metadata.HoodieTableMetadataUtil;
+import org.apache.hudi.metadata.HoodieTableMetadataWriter;
+import org.apache.hudi.metadata.MetadataPartitionType;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.HoodieTable;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
+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.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class TestIndexActionExecutors {
+
+ private HoodieEngineContext context;
+ private HoodieTable table;
+ private HoodieTableMetaClient metaClient;
+ private HoodieTableConfig tableConfig;
+ private HoodieActiveTimeline activeTimeline;
+ private HoodieStorage storage;
+
+ @BeforeEach
+ void setUp() {
+ context = mock(HoodieEngineContext.class);
+ table = mock(HoodieTable.class);
+ metaClient = mock(HoodieTableMetaClient.class);
+ tableConfig = mock(HoodieTableConfig.class);
+ activeTimeline = mock(HoodieActiveTimeline.class);
+ storage = mock(HoodieStorage.class);
+ when(table.getMetaClient()).thenReturn(metaClient);
+ when(table.getActiveTimeline()).thenReturn(activeTimeline);
+ when(table.getInstantGenerator()).thenReturn(INSTANT_GENERATOR);
+ when(table.getStorage()).thenReturn(storage);
+ doReturn(getDefaultStorageConf()).when(storage).getConf();
+ doReturn(getDefaultStorageConf()).when(context).getStorageConf();
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+ when(metaClient.getInstantGenerator()).thenReturn(INSTANT_GENERATOR);
+ when(metaClient.getActiveTimeline()).thenReturn(activeTimeline);
+ when(metaClient.reloadActiveTimeline()).thenReturn(activeTimeline);
+
when(tableConfig.getMetadataPartitions()).thenReturn(Collections.emptySet());
+
when(tableConfig.getMetadataPartitionsInflight()).thenReturn(Collections.emptySet());
+ }
+
+ @Test
+ void testScheduleRejectsSingleWriterConfiguration() {
+ HoodieWriteConfig config =
HoodieWriteConfig.newBuilder().withPath("/table").build();
+ ScheduleIndexActionExecutor executor = new ScheduleIndexActionExecutor(
+ context, config, table, "002",
Collections.singletonList(MetadataPartitionType.COLUMN_STATS),
Collections.emptyList());
+
+ HoodieIndexException exception = assertThrows(HoodieIndexException.class,
executor::execute);
+
assertTrue(exception.getMessage().contains(HoodieWriteConfig.WRITE_CONCURRENCY_MODE.key()));
+ }
+
+ @Test
+ void testScheduleCreatesPlanAndIsIdempotent() {
+ HoodieWriteConfig config = multiWriterConfig();
+ HoodieInstant completed = INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "001");
+ HoodieTimeline completedTimeline = mock(HoodieTimeline.class);
+
when(activeTimeline.getContiguousCompletedWriteTimeline()).thenReturn(completedTimeline);
+ when(completedTimeline.lastInstant()).thenReturn(Option.of(completed));
+ ScheduleIndexActionExecutor executor = new ScheduleIndexActionExecutor(
+ context, config, table, "002",
Collections.singletonList(MetadataPartitionType.COLUMN_STATS),
Collections.emptyList());
+
+ Option<HoodieIndexPlan> plan = executor.execute();
+
+ assertTrue(plan.isPresent());
+ assertEquals(1, plan.get().getIndexPartitionInfos().size());
+ assertEquals(MetadataPartitionType.COLUMN_STATS.getPartitionPath(),
+ plan.get().getIndexPartitionInfos().get(0).getMetadataPartitionPath());
+ assertEquals("001",
plan.get().getIndexPartitionInfos().get(0).getIndexUptoInstant());
+ verify(activeTimeline).saveToPendingIndexAction(any(HoodieInstant.class),
any(HoodieIndexPlan.class));
+
+
when(tableConfig.getMetadataPartitions()).thenReturn(Set.of(MetadataPartitionType.COLUMN_STATS.getPartitionPath()));
+ assertFalse(executor.execute().isPresent());
+ }
+
+ @Test
+ void testScheduleAbortsWhenPendingPlanCannotBeSaved() {
+ HoodieWriteConfig config = multiWriterConfig();
+ HoodieInstant completed = INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "001");
+ HoodieTimeline completedTimeline = mock(HoodieTimeline.class);
+
when(activeTimeline.getContiguousCompletedWriteTimeline()).thenReturn(completedTimeline);
+ when(completedTimeline.lastInstant()).thenReturn(Option.of(completed));
+ doThrow(new HoodieIOException("write failed"))
+
.when(activeTimeline).saveToPendingIndexAction(any(HoodieInstant.class),
any(HoodieIndexPlan.class));
+
+ try (MockedStatic<HoodieTableMetadataUtil> metadataUtil =
mockStatic(HoodieTableMetadataUtil.class)) {
+ metadataUtil.when(() ->
HoodieTableMetadataUtil.getInflightAndCompletedMetadataPartitions(tableConfig))
+ .thenReturn(Collections.emptySet());
+ metadataUtil.when(() -> HoodieTableMetadataUtil.metadataPartitionExists(
+ metaClient.getBasePath(), context,
MetadataPartitionType.COLUMN_STATS.getPartitionPath())).thenReturn(false);
+
+ Option<HoodieIndexPlan> result = new ScheduleIndexActionExecutor(
+ context, config, table, "002",
Collections.singletonList(MetadataPartitionType.COLUMN_STATS),
Collections.emptyList()).execute();
+
+ assertFalse(result.isPresent());
+
verify(activeTimeline).deleteInstantFileIfExists(INSTANT_GENERATOR.getIndexRequestedInstant("002"));
+ }
+ }
+
+ @Test
+ void testRunRejectsInvalidConfigurationAndMissingInstant() {
+ RunIndexActionExecutor invalidExecutor = new RunIndexActionExecutor(
+ context, HoodieWriteConfig.newBuilder().withPath("/table").build(),
table, "002");
+ assertThrows(HoodieIndexException.class, invalidExecutor::execute);
+
+ HoodieWriteConfig config = multiWriterConfig();
+
when(activeTimeline.filterPendingIndexTimeline()).thenReturn(activeTimeline);
+ when(activeTimeline.filter(any())).thenReturn(activeTimeline);
+ when(activeTimeline.lastInstant()).thenReturn(Option.empty());
+ assertThrows(HoodieIndexException.class,
+ () -> new RunIndexActionExecutor(context, config, table,
"002").execute());
+ }
+
+ @Test
+ void testRunRejectsUnreadableAndEmptyPlans() throws IOException {
+ HoodieWriteConfig config = multiWriterConfig();
+ HoodieInstant requested = requestedIndexInstant();
+ stubRequestedIndexInstant(requested);
+ doThrow(new IOException("read
failed")).when(activeTimeline).readIndexPlan(requested);
+ assertThrows(HoodieIndexException.class,
+ () -> new RunIndexActionExecutor(context, config, table,
"002").execute());
+
+ doReturn(new HoodieIndexPlan(1,
Collections.emptyList())).when(activeTimeline).readIndexPlan(requested);
+ assertThrows(HoodieIndexException.class,
+ () -> new RunIndexActionExecutor(context, config, table,
"002").execute());
+ }
+
+ @Test
+ void testRunRejectsPartitionThatAlreadyExists() throws IOException {
+ HoodieWriteConfig config = multiWriterConfig();
+ HoodieInstant requested = requestedIndexInstant();
+ stubRequestedIndexInstant(requested);
+ HoodieIndexPartitionInfo info = new HoodieIndexPartitionInfo(
+ 1, MetadataPartitionType.COLUMN_STATS.getPartitionPath(), "001",
Collections.emptyMap());
+ when(activeTimeline.readIndexPlan(requested)).thenReturn(new
HoodieIndexPlan(1, Collections.singletonList(info)));
+
when(tableConfig.getMetadataPartitions()).thenReturn(Set.of(MetadataPartitionType.COLUMN_STATS.getPartitionPath()));
+
+ assertThrows(HoodieIndexException.class,
+ () -> new RunIndexActionExecutor(context, config, table,
"002").execute());
+ }
+
+ @Test
+ void testRunInitializesFilesPartitionAndCompletesIndexInstant() throws
Exception {
+ HoodieWriteConfig config = multiWriterConfig();
+ HoodieInstant requested = requestedIndexInstant();
+ stubRequestedIndexInstant(requested);
+ HoodieIndexPartitionInfo info = new HoodieIndexPartitionInfo(
+ 1, MetadataPartitionType.FILES.getPartitionPath(), "001",
Collections.emptyMap());
+ when(activeTimeline.readIndexPlan(requested)).thenReturn(new
HoodieIndexPlan(1, Collections.singletonList(info)));
+ HoodieTableMetadataWriter writer = mock(HoodieTableMetadataWriter.class);
+ when(table.getIndexingMetadataWriter("002")).thenReturn(Option.of(writer));
+
+ Option<HoodieIndexCommitMetadata> result;
+ try (MockedConstruction<TransactionManager> ignored =
mockConstruction(TransactionManager.class)) {
+ result = new RunIndexActionExecutor(context, config, table,
"002").execute();
+ }
+
+ assertTrue(result.isPresent());
+ List<HoodieIndexPartitionInfo> completedPartitions =
result.get().getIndexPartitionInfos();
+ assertEquals(1, completedPartitions.size());
+ assertEquals("001", completedPartitions.get(0).getIndexUptoInstant());
+ verify(activeTimeline).transitionIndexRequestedToInflight(requested);
+ verify(tableConfig).setMetadataPartitionState(metaClient,
MetadataPartitionType.FILES.getPartitionPath(), true);
+ verify(activeTimeline).saveAsComplete(any(Boolean.class),
any(HoodieInstant.class), any(Option.class));
+ verify(writer).close();
+ }
+
+ @Test
+ void testRunBuildsPartitionAndCatchesUpConcurrentCommits() throws Exception {
+ HoodieWriteConfig config = multiWriterConfig();
+ HoodieInstant requested = requestedIndexInstant();
+ stubRequestedIndexInstant(requested);
+ HoodieIndexPartitionInfo info = new HoodieIndexPartitionInfo(
+ 1, MetadataPartitionType.COLUMN_STATS.getPartitionPath(), "001",
Collections.emptyMap());
+ when(activeTimeline.readIndexPlan(requested)).thenReturn(new
HoodieIndexPlan(1, Collections.singletonList(info)));
+ HoodieTableMetadataWriter writer = mock(HoodieTableMetadataWriter.class);
+ when(table.getIndexingMetadataWriter("002")).thenReturn(Option.of(writer));
+ when(table.getConfig()).thenReturn(config);
+ when(metaClient.getBasePath()).thenReturn(new StoragePath("/table"));
+
+ HoodieTimeline emptyTimeline = mock(HoodieTimeline.class);
+
when(emptyTimeline.filterInflightsAndRequested()).thenReturn(emptyTimeline);
+ when(emptyTimeline.findInstantsBefore("001")).thenReturn(emptyTimeline);
+ when(emptyTimeline.firstInstant()).thenReturn(Option.empty());
+ when(emptyTimeline.filterCompletedInstants()).thenReturn(emptyTimeline);
+ when(emptyTimeline.getInstantsAsStream()).thenAnswer(ignored ->
java.util.stream.Stream.empty());
+ when(activeTimeline.getTimelineOfActions(any())).thenReturn(emptyTimeline);
+ HoodieInstant completedCommit = INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "002");
+ HoodieTimeline catchupTimeline = mock(HoodieTimeline.class);
+ when(catchupTimeline.getInstantsAsStream()).thenAnswer(ignored ->
java.util.stream.Stream.of(completedCommit));
+ when(activeTimeline.findInstantsAfter("001")).thenReturn(catchupTimeline);
+ HoodieArchivedTimeline archivedTimeline =
mock(HoodieArchivedTimeline.class);
+ when(archivedTimeline.getInstantsAsStream()).thenAnswer(ignored ->
java.util.stream.Stream.empty());
+ when(metaClient.getArchivedTimeline()).thenReturn(archivedTimeline);
+ HoodieCommitMetadata commitMetadata = mock(HoodieCommitMetadata.class);
+
when(activeTimeline.readCommitMetadata(completedCommit)).thenReturn(commitMetadata);
+
+ HoodieTableMetaClient metadataMetaClient =
mock(HoodieTableMetaClient.class);
+ HoodieArchivedTimeline metadataArchivedTimeline =
mock(HoodieArchivedTimeline.class);
+
when(metadataArchivedTimeline.filterCompletedInstants()).thenReturn(metadataArchivedTimeline);
+
when(metadataArchivedTimeline.findInstantsAfter("001")).thenReturn(metadataArchivedTimeline);
+ when(metadataArchivedTimeline.getInstantsAsStream()).thenAnswer(ignored ->
java.util.stream.Stream.empty());
+
when(metadataMetaClient.getArchivedTimeline()).thenReturn(metadataArchivedTimeline);
+ HoodieActiveTimeline metadataActiveTimeline =
mock(HoodieActiveTimeline.class);
+
when(metadataActiveTimeline.filterCompletedInstants()).thenReturn(metadataActiveTimeline);
+
when(metadataActiveTimeline.findInstantsAfter("001")).thenReturn(metadataActiveTimeline);
+ when(metadataActiveTimeline.getInstantsAsStream()).thenAnswer(ignored ->
java.util.stream.Stream.empty());
+
when(metadataActiveTimeline.filter(any())).thenReturn(metadataActiveTimeline);
+ when(metadataActiveTimeline.firstInstant()).thenReturn(Option.empty());
+
when(metadataMetaClient.reloadActiveTimeline()).thenReturn(metadataActiveTimeline);
+ HoodieTableMetaClient.Builder builder =
mock(HoodieTableMetaClient.Builder.class, Mockito.RETURNS_SELF);
+ when(builder.build()).thenReturn(metadataMetaClient);
+
+ Option<HoodieIndexCommitMetadata> result;
+ try (MockedStatic<HoodieTableMetaClient> metaClientStatic =
mockStatic(HoodieTableMetaClient.class);
+ MockedConstruction<TransactionManager> ignoredTxn =
mockConstruction(TransactionManager.class);
+ MockedConstruction<HoodieHeartbeatClient> ignoredHeartbeat =
mockConstruction(HoodieHeartbeatClient.class)) {
+
metaClientStatic.when(HoodieTableMetaClient::builder).thenReturn(builder);
+
+ result = new RunIndexActionExecutor(context, config, table,
"002").execute();
+ }
+
+ assertTrue(result.isPresent());
+ assertEquals("002",
result.get().getIndexPartitionInfos().get(0).getIndexUptoInstant());
+ verify(writer).buildMetadataPartitions(context,
Collections.singletonList(info), "002");
+ verify(writer).update(commitMetadata, "002");
+ verify(writer).close();
+ }
+
+ @Test
+ void testRunAbortsAndCleansPartialIndexOnTimelineFailure() throws Exception {
+ HoodieWriteConfig config = multiWriterConfig();
+ HoodieInstant requested = requestedIndexInstant();
+ stubRequestedIndexInstant(requested);
+ String partition = MetadataPartitionType.FILES.getPartitionPath();
+ HoodieIndexPartitionInfo info = new HoodieIndexPartitionInfo(1, partition,
"001", Collections.emptyMap());
+ when(activeTimeline.readIndexPlan(requested)).thenReturn(new
HoodieIndexPlan(1, Collections.singletonList(info)));
+ HoodieTableMetadataWriter writer = mock(HoodieTableMetadataWriter.class);
+ when(table.getIndexingMetadataWriter("002")).thenReturn(Option.of(writer));
+ doAnswer(ignored -> {
+ throw new IOException("timeline failure");
+ })
+ .when(activeTimeline).saveAsComplete(any(Boolean.class),
any(HoodieInstant.class), any(Option.class));
+ when(tableConfig.getMetadataPartitionsInflight()).thenReturn(new
HashSet<>(Collections.singleton(partition)));
+ when(tableConfig.getMetadataPartitions()).thenReturn(new
HashSet<>(Collections.singleton(partition)));
+ when(metaClient.getBasePath()).thenReturn(new StoragePath("/table"));
+
+ try (MockedStatic<HoodieTableConfig> tableConfigStatic =
mockStatic(HoodieTableConfig.class);
+ MockedStatic<HoodieTableMetadataUtil> metadataUtil =
mockStatic(HoodieTableMetadataUtil.class);
+ MockedConstruction<TransactionManager> ignored =
mockConstruction(TransactionManager.class)) {
+ metadataUtil.when(() -> HoodieTableMetadataUtil.metadataPartitionExists(
+ metaClient.getBasePath(), context, partition)).thenReturn(true);
+
+ assertThrows(HoodieIndexException.class,
+ () -> new RunIndexActionExecutor(context, config, table,
"002").execute());
+
+ metadataUtil.verify(() ->
HoodieTableMetadataUtil.deleteMetadataPartition(metaClient.getBasePath(),
context, partition));
+ tableConfigStatic.verify(() -> HoodieTableConfig.update(storage,
metaClient.getMetaPath(), tableConfig.getProps()));
+ }
+ verify(writer).close();
+
verify(activeTimeline).deleteInstantFileIfExists(INSTANT_GENERATOR.getIndexInflightInstant("002"));
+ }
+
+ private HoodieWriteConfig multiWriterConfig() {
+ return HoodieWriteConfig.newBuilder()
+ .withPath("/table")
+
.withWriteConcurrencyMode(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL)
+
.withLockConfig(HoodieLockConfig.newBuilder().withLockProvider(InProcessLockProvider.class).build())
+ .build();
+ }
+
+ private HoodieInstant requestedIndexInstant() {
+ return INSTANT_GENERATOR.getIndexRequestedInstant("002");
+ }
+
+ private void stubRequestedIndexInstant(HoodieInstant requested) {
+
when(activeTimeline.filterPendingIndexTimeline()).thenReturn(activeTimeline);
+ when(activeTimeline.filter(any())).thenReturn(activeTimeline);
+ when(activeTimeline.lastInstant()).thenReturn(Option.of(requested));
+ }
+}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/index/TestIndexingCatchupTask.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/index/TestIndexingCatchupTask.java
index ca34060694df..5d90b7479c24 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/index/TestIndexingCatchupTask.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/index/TestIndexingCatchupTask.java
@@ -19,6 +19,9 @@
package org.apache.hudi.table.action.index;
+import org.apache.hudi.avro.model.HoodieCleanMetadata;
+import org.apache.hudi.avro.model.HoodieRestoreMetadata;
+import org.apache.hudi.avro.model.HoodieRollbackMetadata;
import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient;
import org.apache.hudi.client.transaction.TransactionManager;
import org.apache.hudi.common.engine.HoodieEngineContext;
@@ -26,6 +29,8 @@ import
org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.CleanerUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieCleanConfig;
import org.apache.hudi.config.HoodieWriteConfig;
@@ -38,6 +43,7 @@ import org.apache.hudi.table.HoodieTable;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mock;
+import org.mockito.MockedStatic;
import org.mockito.MockitoAnnotations;
import java.io.IOException;
@@ -50,10 +56,13 @@ import java.util.concurrent.atomic.AtomicInteger;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class TestIndexingCatchupTask {
@@ -240,6 +249,120 @@ public class TestIndexingCatchupTask {
assertTrue(task.awaitInstantCaughtUp(pendingInstantWithNoHeartbeat),
"Expected null as the instant's heartbeat has expired.");
}
+ @Test
+ public void testRunSkipsInstantAlreadyCommittedToMetadata() {
+ HoodieInstant instant =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.COMMIT_ACTION, "002");
+ HoodieActiveTimeline metadataTimeline = mock(HoodieActiveTimeline.class);
+
when(metadataMetaClient.reloadActiveTimeline()).thenReturn(metadataTimeline);
+
when(metadataTimeline.filterCompletedInstants()).thenReturn(metadataTimeline);
+ when(metadataTimeline.filter(any())).thenReturn(metadataTimeline);
+ when(metadataTimeline.firstInstant()).thenReturn(Option.of(instant));
+
+ RunningIndexingCatchupTask task = runningTask(instant);
+ task.run();
+
+ assertEquals("002", task.currentCaughtupInstant);
+ assertEquals(0, task.writeActionsUpdated.get());
+ }
+
+ @Test
+ public void testRunUpdatesCompletedWriteActionInsideTransaction() {
+ HoodieInstant instant =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.REPLACE_COMMIT_ACTION, "002");
+ stubNoCompletedMetadataInstant();
+ RunningIndexingCatchupTask task = runningTask(instant);
+
+ task.run();
+
+ assertEquals(1, task.writeActionsUpdated.get());
+ assertEquals("002", task.currentCaughtupInstant);
+ verify(transactionManager).beginStateChange(Option.of(instant),
Option.empty());
+ verify(transactionManager).endStateChange(Option.of(instant));
+ }
+
+ @Test
+ public void testRunUpdatesCleanRestoreAndRollbackActions() throws
IOException {
+ HoodieInstant clean =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.CLEAN_ACTION, "002");
+ HoodieInstant restore =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.RESTORE_ACTION, "003");
+ HoodieInstant rollback =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.ROLLBACK_ACTION, "004");
+ stubNoCompletedMetadataInstant();
+ HoodieCleanMetadata cleanMetadata = mock(HoodieCleanMetadata.class);
+ HoodieRestoreMetadata restoreMetadata = mock(HoodieRestoreMetadata.class);
+ HoodieRollbackMetadata rollbackMetadata =
mock(HoodieRollbackMetadata.class);
+ HoodieActiveTimeline activeTimeline = mock(HoodieActiveTimeline.class);
+ when(metaClient.getActiveTimeline()).thenReturn(activeTimeline);
+
when(activeTimeline.readRestoreMetadata(restore)).thenReturn(restoreMetadata);
+
when(activeTimeline.readRollbackMetadata(rollback)).thenReturn(rollbackMetadata);
+
+ try (MockedStatic<CleanerUtils> cleanerUtils =
mockStatic(CleanerUtils.class)) {
+ cleanerUtils.when(() -> CleanerUtils.getCleanerMetadata(metaClient,
clean)).thenReturn(cleanMetadata);
+ runningTask(clean, restore, rollback).run();
+ }
+
+ verify(metadataWriter).update(cleanMetadata, "002");
+ verify(metadataWriter).update(restoreMetadata, "003");
+ verify(metadataWriter).update(rollbackMetadata, "004");
+ verify(transactionManager).endStateChange(Option.of(clean));
+ verify(transactionManager).endStateChange(Option.of(restore));
+ verify(transactionManager).endStateChange(Option.of(rollback));
+ }
+
+ @Test
+ public void testRunRejectsUnexpectedCompletedActionAndReleasesTransaction() {
+ HoodieInstant instant =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.SAVEPOINT_ACTION, "002");
+ stubNoCompletedMetadataInstant();
+
+ assertThrows(IllegalStateException.class, () ->
runningTask(instant).run());
+ verify(transactionManager).endStateChange(Option.of(instant));
+ }
+
+ @Test
+ public void testAwaitInstantCaughtUpUsesKnownMetadataInstant() {
+ HoodieInstant instant =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.INFLIGHT,
HoodieTimeline.COMMIT_ACTION, "002");
+ RunningIndexingCatchupTask task = new RunningIndexingCatchupTask(
+ metadataWriter, Collections.singletonList(instant),
Collections.singleton("002"), metaClient, metadataMetaClient,
+ transactionManager, "001", engineContext, table, heartbeatClient);
+
+ assertTrue(task.awaitInstantCaughtUp(instant));
+ assertEquals("002", task.currentCaughtupInstant);
+ }
+
+ private RunningIndexingCatchupTask runningTask(HoodieInstant... instants) {
+ return new RunningIndexingCatchupTask(
+ metadataWriter, java.util.Arrays.asList(instants), new HashSet<>(),
metaClient, metadataMetaClient,
+ transactionManager, "001", engineContext, table, heartbeatClient);
+ }
+
+ private void stubNoCompletedMetadataInstant() {
+ HoodieActiveTimeline metadataTimeline = mock(HoodieActiveTimeline.class);
+
when(metadataMetaClient.reloadActiveTimeline()).thenReturn(metadataTimeline);
+
when(metadataTimeline.filterCompletedInstants()).thenReturn(metadataTimeline);
+ when(metadataTimeline.filter(any())).thenReturn(metadataTimeline);
+ when(metadataTimeline.firstInstant()).thenReturn(Option.empty());
+ }
+
+ static class RunningIndexingCatchupTask extends AbstractIndexingCatchupTask {
+ private final AtomicInteger writeActionsUpdated = new AtomicInteger();
+
+ RunningIndexingCatchupTask(HoodieTableMetadataWriter metadataWriter,
+ List<HoodieInstant> instantsToIndex,
+ Set<String> metadataCompletedInstants,
+ HoodieTableMetaClient metaClient,
+ HoodieTableMetaClient metadataMetaClient,
+ TransactionManager transactionManager,
+ String currentCaughtupInstant,
+ HoodieEngineContext engineContext,
+ HoodieTable table,
+ HoodieHeartbeatClient heartbeatClient) {
+ super(metadataWriter, instantsToIndex, metadataCompletedInstants,
metaClient, metadataMetaClient,
+ transactionManager, currentCaughtupInstant, engineContext, table,
heartbeatClient);
+ }
+
+ @Override
+ public void updateIndexForWriteAction(HoodieInstant instant) {
+ writeActionsUpdated.incrementAndGet();
+ }
+ }
+
static class DummyIndexingCatchupTask extends AbstractIndexingCatchupTask {
public DummyIndexingCatchupTask(HoodieTableMetadataWriter metadataWriter,
List<HoodieInstant> instantsToIndex,
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestLegacyUpgradeDowngradeHandlers.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestLegacyUpgradeDowngradeHandlers.java
new file mode 100644
index 000000000000..59f855a7161e
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestLegacyUpgradeDowngradeHandlers.java
@@ -0,0 +1,354 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.table.upgrade;
+
+import org.apache.hudi.common.HoodieRollbackStat;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.IOType;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.marker.MarkerType;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.InstantFileNameGenerator;
+import org.apache.hudi.common.util.HoodieStorageUtils;
+import org.apache.hudi.common.util.MarkerUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieRollbackException;
+import org.apache.hudi.exception.HoodieUpgradeDowngradeException;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.marker.DirectWriteMarkers;
+import org.apache.hudi.table.marker.WriteMarkers;
+import org.apache.hudi.table.marker.WriteMarkersFactory;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
+import static
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
+import static
org.apache.hudi.common.util.PartitionPathEncodeUtils.DEPRECATED_DEFAULT_PARTITION_PATH;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class TestLegacyUpgradeDowngradeHandlers {
+
+ @Test
+ void testZeroToOneRecreatesMarkersAndSkipsCurrentInstant() {
+ HoodieTable table = mockTableWithPendingInstants("001", "002");
+ HoodieEngineContext context = mock(HoodieEngineContext.class);
+ doReturn(getDefaultStorageConf()).when(context).getStorageConf();
+ HoodieWriteConfig config = mock(HoodieWriteConfig.class);
+ SupportsUpgradeDowngrade helper = mock(SupportsUpgradeDowngrade.class);
+ when(helper.getTable(config, context)).thenReturn(table);
+ when(config.getMarkersDeleteParallelism()).thenReturn(3);
+
+ ZeroToOneUpgradeHandler handler = spy(new ZeroToOneUpgradeHandler());
+ doNothing().when(handler).recreateMarkers(anyString(), eq(table),
eq(context), anyInt());
+
+ handler.upgrade(config, context, "002", helper);
+
+ verify(handler).recreateMarkers("001", table, context, 3);
+ verify(handler, never()).recreateMarkers("002", table, context, 3);
+ }
+
+ @Test
+ void testZeroToOneCreatesBaseAndLogMarkers() {
+ HoodieInstant instant =
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.INFLIGHT,
HoodieTimeline.DELTA_COMMIT_ACTION, "001");
+ HoodieTable table = mock(HoodieTable.class);
+ HoodieActiveTimeline activeTimeline = mock(HoodieActiveTimeline.class);
+ HoodieTimeline commitsTimeline = mock(HoodieTimeline.class);
+ when(table.getActiveTimeline()).thenReturn(activeTimeline);
+ when(activeTimeline.getCommitsTimeline()).thenReturn(commitsTimeline);
+ when(commitsTimeline.getInstantsAsStream()).thenReturn(Stream.of(instant));
+ when(table.getBaseFileExtension()).thenReturn(".parquet");
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ when(table.getMetaClient()).thenReturn(metaClient);
+ when(metaClient.getBasePath()).thenReturn(new StoragePath("/table"));
+
+ StoragePathInfo logFile = new StoragePathInfo(
+ new StoragePath("partition/.file-id_001.log.1_1-0-1"), 1L, false,
(short) 1, 1L, 1L);
+ HoodieRollbackStat rollbackStat = HoodieRollbackStat.newBuilder()
+ .withPartitionPath("partition")
+ .withDeletedFileResult("/table/partition/file.parquet", true)
+ .withRollbackBlockAppendResults(Collections.singletonMap(logFile, 1L))
+ .build();
+ ZeroToOneUpgradeHandler handler = new ZeroToOneUpgradeHandler() {
+ @Override
+ List<HoodieRollbackStat> getListBasedRollBackStats(
+ HoodieTable<?, ?, ?, ?> ignoredTable, HoodieEngineContext
ignoredContext, Option<HoodieInstant> ignoredInstant) {
+ return Collections.singletonList(rollbackStat);
+ }
+ };
+
+ WriteMarkers markers = mock(WriteMarkers.class);
+ try (MockedStatic<WriteMarkersFactory> markerFactory =
mockStatic(WriteMarkersFactory.class)) {
+ markerFactory.when(() -> WriteMarkersFactory.get(MarkerType.DIRECT,
table, "001")).thenReturn(markers);
+
+ handler.recreateMarkers("001", table, mock(HoodieEngineContext.class),
2);
+
+ verify(markers).quietDeleteMarkerDir(any(HoodieEngineContext.class),
eq(2));
+ verify(markers).create("partition", "file.parquet", IOType.MERGE);
+ verify(markers).create("partition", "file-id_1-0-1_001.parquet",
IOType.APPEND);
+ }
+ }
+
+ @Test
+ void testZeroToOneIgnoresMissingInstantAndWrapsFailures() {
+ HoodieTable table = mock(HoodieTable.class);
+ HoodieActiveTimeline activeTimeline = mock(HoodieActiveTimeline.class);
+ HoodieTimeline timeline = mock(HoodieTimeline.class);
+ when(table.getActiveTimeline()).thenReturn(activeTimeline);
+ when(activeTimeline.getCommitsTimeline()).thenReturn(timeline);
+ when(timeline.getInstantsAsStream()).thenReturn(Stream.empty());
+ assertDoesNotThrow(() -> new
ZeroToOneUpgradeHandler().recreateMarkers("001", table,
mock(HoodieEngineContext.class), 1));
+
+ when(timeline.getInstantsAsStream()).thenThrow(new
RuntimeException("timeline failure"));
+ assertThrows(HoodieRollbackException.class,
+ () -> new ZeroToOneUpgradeHandler().recreateMarkers("001", table,
mock(HoodieEngineContext.class), 1));
+ }
+
+ @Test
+ void testTwoToOneConvertsTimelineServerMarkersToDirectMarkers() throws
Exception {
+ HoodieTable table = mockTableWithPendingInstants("001");
+ HoodieTableMetaClient metaClient = table.getMetaClient();
+
when(metaClient.getMarkerFolderPath("001")).thenReturn("/table/.hoodie/.temp/001");
+ HoodieEngineContext context = mock(HoodieEngineContext.class);
+ doReturn(getDefaultStorageConf()).when(context).getStorageConf();
+ HoodieWriteConfig config = mock(HoodieWriteConfig.class);
+ when(config.getMarkersDeleteParallelism()).thenReturn(2);
+ SupportsUpgradeDowngrade helper = mock(SupportsUpgradeDowngrade.class);
+ when(helper.getTable(config, context)).thenReturn(table);
+ HoodieStorage storage = mock(HoodieStorage.class);
+ Map<String, Set<String>> markerMap = Collections.singletonMap("MARKERS0",
+ Set.of("partition/file.parquet.marker.CREATE",
"partition/file2.parquet.marker.MERGE"));
+
+ try (MockedStatic<HoodieStorageUtils> storageUtils =
mockStatic(HoodieStorageUtils.class);
+ MockedStatic<MarkerUtils> markerUtils = mockStatic(MarkerUtils.class);
+ MockedStatic<org.apache.hudi.common.fs.FSUtils> fsUtils =
mockStatic(org.apache.hudi.common.fs.FSUtils.class);
+ MockedConstruction<DirectWriteMarkers> directMarkers =
mockConstruction(DirectWriteMarkers.class)) {
+ storageUtils.when(() ->
HoodieStorageUtils.getStorage(eq("/table/.hoodie/.temp/001"),
any())).thenReturn(storage);
+ markerUtils.when(() -> MarkerUtils.readMarkerType(storage,
"/table/.hoodie/.temp/001"))
+ .thenReturn(Option.of(MarkerType.TIMELINE_SERVER_BASED));
+ markerUtils.when(() ->
MarkerUtils.readTimelineServerBasedMarkersFromFileSystem(
+ "/table/.hoodie/.temp/001", storage, context,
2)).thenReturn(markerMap);
+
+ new TwoToOneDowngradeHandler().downgrade(config, context, null, helper);
+
+ DirectWriteMarkers direct = directMarkers.constructed().get(0);
+ verify(direct).create("partition/file.parquet.marker.CREATE");
+ verify(direct).create("partition/file2.parquet.marker.MERGE");
+ markerUtils.verify(() -> MarkerUtils.deleteMarkerTypeFile(storage,
"/table/.hoodie/.temp/001"));
+ fsUtils.verify(() ->
org.apache.hudi.common.fs.FSUtils.parallelizeSubPathProcess(
+ eq(context), eq(storage), eq(new
StoragePath("/table/.hoodie/.temp/001")), eq(2), any(), any()));
+ }
+ }
+
+ @Test
+ void testTwoToOneCleansPartialMarkersAndRejectsUnsupportedMarkerType()
throws Exception {
+ HoodieTable table = mockTableWithPendingInstants("001");
+
when(table.getMetaClient().getMarkerFolderPath("001")).thenReturn("/markers/001");
+ HoodieEngineContext context = mock(HoodieEngineContext.class);
+ doReturn(getDefaultStorageConf()).when(context).getStorageConf();
+ HoodieWriteConfig config = mock(HoodieWriteConfig.class);
+ when(config.getMarkersDeleteParallelism()).thenReturn(1);
+ SupportsUpgradeDowngrade helper = mock(SupportsUpgradeDowngrade.class);
+ when(helper.getTable(config, context)).thenReturn(table);
+ HoodieStorage storage = mock(HoodieStorage.class);
+ when(storage.exists(new StoragePath("/markers/001"))).thenReturn(true);
+
+ try (MockedStatic<HoodieStorageUtils> storageUtils =
mockStatic(HoodieStorageUtils.class);
+ MockedStatic<MarkerUtils> markerUtils = mockStatic(MarkerUtils.class);
+ MockedStatic<org.apache.hudi.common.fs.FSUtils> fsUtils =
mockStatic(org.apache.hudi.common.fs.FSUtils.class)) {
+ storageUtils.when(() ->
HoodieStorageUtils.getStorage(eq("/markers/001"), any())).thenReturn(storage);
+ markerUtils.when(() -> MarkerUtils.readMarkerType(storage,
"/markers/001")).thenReturn(Option.empty());
+
+ new TwoToOneDowngradeHandler().downgrade(config, context, null, helper);
+ fsUtils.verify(() ->
org.apache.hudi.common.fs.FSUtils.parallelizeSubPathProcess(
+ eq(context), eq(storage), eq(new StoragePath("/markers/001")),
eq(1), any(), any()));
+
+ markerUtils.when(() -> MarkerUtils.readMarkerType(storage,
"/markers/001")).thenReturn(Option.of(MarkerType.DIRECT));
+ assertThrows(HoodieException.class,
+ () -> new TwoToOneDowngradeHandler().downgrade(config, context,
null, helper));
+ }
+ }
+
+ @Test
+ void testFourToFiveValidatesDefaultPartitionLayouts() throws Exception {
+ HoodieTable table = mock(HoodieTable.class);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+ HoodieStorage storage = mock(HoodieStorage.class);
+ when(table.getMetaClient()).thenReturn(metaClient);
+ when(table.getStorage()).thenReturn(storage);
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+ when(tableConfig.isTablePartitioned()).thenReturn(true);
+ when(tableConfig.getHiveStylePartitioningEnable()).thenReturn("false");
+ when(storage.exists(new StoragePath("/table/" +
DEPRECATED_DEFAULT_PARTITION_PATH))).thenReturn(true);
+ HoodieWriteConfig config = mock(HoodieWriteConfig.class);
+ when(config.getBasePath()).thenReturn("/table");
+ SupportsUpgradeDowngrade helper = mock(SupportsUpgradeDowngrade.class);
+ HoodieEngineContext context = mock(HoodieEngineContext.class);
+ when(helper.getTable(config, context)).thenReturn(table);
+
+ assertThrows(HoodieException.class,
+ () -> new FourToFiveUpgradeHandler().upgrade(config, context, null,
helper));
+
+ when(tableConfig.getHiveStylePartitioningEnable()).thenReturn("true");
+ when(tableConfig.getPartitionFields()).thenReturn(Option.of(new String[]
{"dt", "hh"}));
+ when(storage.exists(new StoragePath("/table/dt=" +
DEPRECATED_DEFAULT_PARTITION_PATH))).thenReturn(false);
+ assertDoesNotThrow(() -> new FourToFiveUpgradeHandler().upgrade(config,
context, null, helper));
+
+ when(config.doSkipDefaultPartitionValidation()).thenReturn(true);
+ assertDoesNotThrow(() -> new FourToFiveUpgradeHandler().upgrade(config,
context, null, helper));
+ }
+
+ @Test
+ void testFourToFiveHandlesNonPartitionedTableAndStorageFailure() throws
Exception {
+ HoodieTable table = mock(HoodieTable.class);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+ HoodieStorage storage = mock(HoodieStorage.class);
+ when(table.getMetaClient()).thenReturn(metaClient);
+ when(table.getStorage()).thenReturn(storage);
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+ HoodieWriteConfig config = mock(HoodieWriteConfig.class);
+ when(config.getBasePath()).thenReturn("/table");
+ SupportsUpgradeDowngrade helper = mock(SupportsUpgradeDowngrade.class);
+ HoodieEngineContext context = mock(HoodieEngineContext.class);
+ when(helper.getTable(config, context)).thenReturn(table);
+
+ when(tableConfig.isTablePartitioned()).thenReturn(false);
+ assertDoesNotThrow(() -> new FourToFiveUpgradeHandler().upgrade(config,
context, null, helper));
+
+ when(tableConfig.isTablePartitioned()).thenReturn(true);
+ when(tableConfig.getHiveStylePartitioningEnable()).thenReturn("false");
+ when(storage.exists(any(StoragePath.class))).thenThrow(new
IOException("storage failure"));
+ assertThrows(HoodieException.class,
+ () -> new FourToFiveUpgradeHandler().upgrade(config, context, null,
helper));
+ }
+
+ @Test
+ void testFiveToSixDeletesRequestedCompactionFromAuxiliaryFolder() throws
Exception {
+ HoodieTable table = mock(HoodieTable.class);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline activeTimeline = mock(HoodieActiveTimeline.class);
+ HoodieTimeline compactionTimeline = mock(HoodieTimeline.class);
+ InstantFileNameGenerator fileNameGenerator =
mock(InstantFileNameGenerator.class);
+ HoodieStorage storage = mock(HoodieStorage.class);
+ HoodieInstant requested = INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.REQUESTED, HoodieTimeline.COMPACTION_ACTION,
"001");
+ when(table.getMetaClient()).thenReturn(metaClient);
+ when(metaClient.getActiveTimeline()).thenReturn(activeTimeline);
+
when(activeTimeline.filterPendingCompactionTimeline()).thenReturn(compactionTimeline);
+ when(compactionTimeline.filter(any())).thenReturn(compactionTimeline);
+
when(compactionTimeline.getInstantsAsStream()).thenReturn(Stream.of(requested));
+
when(metaClient.getInstantFileNameGenerator()).thenReturn(fileNameGenerator);
+
when(fileNameGenerator.getFileName(requested)).thenReturn("001.compaction.requested");
+ when(metaClient.getMetaAuxiliaryPath()).thenReturn("/table/.hoodie/.aux");
+ when(metaClient.getStorage()).thenReturn(storage);
+ StoragePath auxFile = new
StoragePath("/table/.hoodie/.aux/001.compaction.requested");
+ when(storage.exists(auxFile)).thenReturn(true);
+ HoodieWriteConfig config = mock(HoodieWriteConfig.class);
+ HoodieEngineContext context = mock(HoodieEngineContext.class);
+ SupportsUpgradeDowngrade helper = mock(SupportsUpgradeDowngrade.class);
+ when(helper.getTable(config, context)).thenReturn(table);
+
+ new FiveToSixUpgradeHandler().upgrade(config, context, null, helper);
+ verify(storage).deleteFile(auxFile);
+
+ when(compactionTimeline.getInstantsAsStream()).thenReturn(Stream.empty());
+ assertDoesNotThrow(() -> new FiveToSixUpgradeHandler().upgrade(config,
context, null, helper));
+ }
+
+ @Test
+ void testFiveToSixWrapsAuxiliaryFolderDeleteFailure() throws Exception {
+ HoodieTable table = mock(HoodieTable.class);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline activeTimeline = mock(HoodieActiveTimeline.class);
+ HoodieTimeline compactionTimeline = mock(HoodieTimeline.class);
+ InstantFileNameGenerator fileNameGenerator =
mock(InstantFileNameGenerator.class);
+ HoodieStorage storage = mock(HoodieStorage.class);
+ HoodieInstant requested = INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.REQUESTED, HoodieTimeline.COMPACTION_ACTION,
"001");
+ when(table.getMetaClient()).thenReturn(metaClient);
+ when(metaClient.getActiveTimeline()).thenReturn(activeTimeline);
+
when(activeTimeline.filterPendingCompactionTimeline()).thenReturn(compactionTimeline);
+ when(compactionTimeline.filter(any())).thenReturn(compactionTimeline);
+
when(compactionTimeline.getInstantsAsStream()).thenReturn(Stream.of(requested));
+
when(metaClient.getInstantFileNameGenerator()).thenReturn(fileNameGenerator);
+
when(fileNameGenerator.getFileName(requested)).thenReturn("001.compaction.requested");
+ when(metaClient.getMetaAuxiliaryPath()).thenReturn("/table/.hoodie/.aux");
+ when(metaClient.getStorage()).thenReturn(storage);
+ StoragePath auxFile = new
StoragePath("/table/.hoodie/.aux/001.compaction.requested");
+ when(storage.exists(auxFile)).thenThrow(new IOException("failure"));
+ HoodieWriteConfig config = mock(HoodieWriteConfig.class);
+ HoodieEngineContext context = mock(HoodieEngineContext.class);
+ SupportsUpgradeDowngrade helper = mock(SupportsUpgradeDowngrade.class);
+ when(helper.getTable(config, context)).thenReturn(table);
+
+ assertThrows(HoodieUpgradeDowngradeException.class,
+ () -> new FiveToSixUpgradeHandler().upgrade(config, context, null,
helper));
+ }
+
+ @SuppressWarnings("unchecked")
+ private HoodieTable mockTableWithPendingInstants(String... instantTimes) {
+ HoodieTable table = mock(HoodieTable.class);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline activeTimeline = mock(HoodieActiveTimeline.class);
+ HoodieTimeline pendingTimeline = mock(HoodieTimeline.class);
+ List<HoodieInstant> instants = Arrays.stream(instantTimes)
+ .map(t ->
INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.INFLIGHT,
HoodieTimeline.COMMIT_ACTION, t))
+ .collect(Collectors.toList());
+ when(table.getMetaClient()).thenReturn(metaClient);
+ when(metaClient.getCommitsTimeline()).thenReturn(activeTimeline);
+
when(activeTimeline.filterPendingExcludingCompactionAndLogCompaction()).thenReturn(pendingTimeline);
+ when(pendingTimeline.getReverseOrderedInstants()).thenAnswer(ignored ->
instants.stream());
+ return table;
+ }
+}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngradeOrchestration.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngradeOrchestration.java
new file mode 100644
index 000000000000..150fa996091e
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngradeOrchestration.java
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.table.upgrade;
+
+import org.apache.hudi.common.config.ConfigProperty;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieUpgradeDowngradeException;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.HoodieTable;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+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.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class TestUpgradeDowngradeOrchestration {
+
+ @Test
+ void testUpgradeRunsEveryVersionHopAndRollbackOnce() throws Exception {
+ TestContext testContext = new TestContext(HoodieTableVersion.SIX);
+ TrackingUpgradeDowngrade upgradeDowngrade =
testContext.trackingUpgradeDowngrade();
+
+ try (MockedStatic<UpgradeDowngradeUtils> upgradeUtils =
mockStatic(UpgradeDowngradeUtils.class);
+ MockedStatic<HoodieTableConfig> tableConfigStatic =
mockStatic(HoodieTableConfig.class)) {
+ upgradeDowngrade.run(HoodieTableVersion.NINE, "100");
+
+ assertEquals(Arrays.asList("6->7", "7->8", "8->9"),
upgradeDowngrade.hops);
+ verify(testContext.tableConfig).setTableVersion(HoodieTableVersion.NINE);
+ upgradeUtils.verify(() ->
UpgradeDowngradeUtils.rollbackFailedWritesAndCompact(
+ testContext.table, testContext.engineContext, testContext.config,
testContext.helper, false, HoodieTableVersion.SIX));
+ tableConfigStatic.verify(() -> HoodieTableConfig.updateAndDeleteProps(
+ eq(testContext.storage), any(StoragePath.class), any(),
eq(java.util.Collections.emptySet())));
+ }
+ }
+
+ @Test
+ void testDowngradeRunsEveryVersionHopAndRollbackOnce() throws Exception {
+ TestContext testContext = new TestContext(HoodieTableVersion.NINE);
+ TrackingUpgradeDowngrade upgradeDowngrade =
testContext.trackingUpgradeDowngrade();
+
+ try (MockedStatic<UpgradeDowngradeUtils> upgradeUtils =
mockStatic(UpgradeDowngradeUtils.class);
+ MockedStatic<HoodieTableConfig> ignored =
mockStatic(HoodieTableConfig.class)) {
+ upgradeDowngrade.run(HoodieTableVersion.SIX, null);
+
+ assertEquals(Arrays.asList("9->8", "8->7", "7->6"),
upgradeDowngrade.hops);
+ verify(testContext.tableConfig).setTableVersion(HoodieTableVersion.SIX);
+ upgradeUtils.verify(() ->
UpgradeDowngradeUtils.rollbackFailedWritesAndCompact(
+ testContext.table, testContext.engineContext, testContext.config,
testContext.helper, false, HoodieTableVersion.NINE));
+ }
+ }
+
+ @Test
+ void testHandlerFailureDoesNotPublishTargetVersion() throws Exception {
+ TestContext testContext = new TestContext(HoodieTableVersion.SEVEN);
+ TrackingUpgradeDowngrade upgradeDowngrade =
testContext.trackingUpgradeDowngrade();
+ upgradeDowngrade.failure = new IllegalStateException("handler failed");
+
+ try (MockedStatic<UpgradeDowngradeUtils> upgradeUtils =
mockStatic(UpgradeDowngradeUtils.class);
+ MockedStatic<HoodieTableConfig> tableConfigStatic =
mockStatic(HoodieTableConfig.class)) {
+ assertThrows(IllegalStateException.class,
+ () -> upgradeDowngrade.run(HoodieTableVersion.EIGHT, "100"));
+
+ assertEquals(java.util.Collections.singletonList("7->8"),
upgradeDowngrade.hops);
+ verify(testContext.tableConfig,
never()).setTableVersion(any(HoodieTableVersion.class));
+ tableConfigStatic.verifyNoInteractions();
+ upgradeUtils.verify(() ->
UpgradeDowngradeUtils.rollbackFailedWritesAndCompact(
+ testContext.table, testContext.engineContext, testContext.config,
testContext.helper, false, HoodieTableVersion.SIX));
+ }
+ }
+
+ @Test
+ void testVersionValidationAndNoOpPaths() throws Exception {
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+ HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .withPath("/table")
+ .withAutoUpgradeVersion(false)
+ .build();
+
+ assertThrows(HoodieUpgradeDowngradeException.class,
+ () -> UpgradeDowngrade.needsDowngrade(HoodieTableVersion.SIX,
HoodieTableVersion.FIVE));
+ when(tableConfig.getTableVersion()).thenReturn(HoodieTableVersion.FIVE);
+ assertThrows(HoodieUpgradeDowngradeException.class,
+ () -> UpgradeDowngrade.needsUpgrade(metaClient, config,
HoodieTableVersion.SIX));
+ when(tableConfig.getTableVersion()).thenReturn(HoodieTableVersion.SIX);
+ assertFalse(UpgradeDowngrade.needsUpgrade(metaClient, config,
HoodieTableVersion.NINE));
+ assertEquals(HoodieTableVersion.SIX, config.getWriteVersion());
+
+ TestContext noOpContext = new TestContext(HoodieTableVersion.SIX);
+ noOpContext.trackingUpgradeDowngrade().run(HoodieTableVersion.SIX, null);
+ }
+
+ @Test
+ void testUnsupportedDirectHopsVisitEveryHandlerBranch() throws Exception {
+ TestContext testContext = new TestContext(HoodieTableVersion.SIX);
+ TrackingUpgradeDowngrade upgradeDowngrade =
testContext.trackingUpgradeDowngrade();
+
+ assertThrows(HoodieUpgradeDowngradeException.class,
+ () -> upgradeDowngrade.callRealUpgrade(HoodieTableVersion.SIX,
HoodieTableVersion.NINE));
+ assertThrows(HoodieUpgradeDowngradeException.class,
+ () -> upgradeDowngrade.callRealDowngrade(HoodieTableVersion.NINE,
HoodieTableVersion.SIX));
+ }
+
+ @Test
+ void testHandlerConfigChangesAreAppliedWithAlternativeKeys() throws
Exception {
+ TestContext testContext = new TestContext(HoodieTableVersion.SIX);
+ TrackingUpgradeDowngrade upgradeDowngrade =
testContext.trackingUpgradeDowngrade();
+ ConfigProperty<String> propertyToAdd =
ConfigProperty.key("test.property").noDefaultValue()
+ .withAlternatives("test.property.legacy");
+ ConfigProperty<String> propertyToRemove =
ConfigProperty.key("test.remove").noDefaultValue();
+ upgradeDowngrade.changeSet = new UpgradeDowngrade.TableConfigChangeSet(
+ Map.of(propertyToAdd, "value"), Set.of(propertyToRemove));
+
+ try (MockedStatic<HoodieTableConfig> ignored =
mockStatic(HoodieTableConfig.class)) {
+ upgradeDowngrade.run(HoodieTableVersion.SEVEN, null);
+ }
+
+ verify(testContext.tableConfig).clearValue(propertyToRemove);
+ verify(testContext.tableConfig).setValue(propertyToAdd, "value");
+ verify(testContext.tableConfig).setValue("test.property.legacy", "value");
+ }
+
+ @Test
+ void testMetadataTableLookupFailureIsWrapped() throws Exception {
+ TestContext testContext = new TestContext(HoodieTableVersion.SIX);
+ when(testContext.storage.exists(any(StoragePath.class))).thenThrow(new
java.io.IOException("lookup failed"));
+
+ assertThrows(HoodieUpgradeDowngradeException.class,
+ () ->
testContext.trackingUpgradeDowngrade().run(HoodieTableVersion.SEVEN, null));
+ }
+
+ private static class TestContext {
+ private final HoodieTableMetaClient metaClient =
mock(HoodieTableMetaClient.class);
+ private final HoodieTableConfig tableConfig =
mock(HoodieTableConfig.class);
+ private final HoodieStorage storage = mock(HoodieStorage.class);
+ private final HoodieEngineContext engineContext =
mock(HoodieEngineContext.class);
+ private final SupportsUpgradeDowngrade helper =
mock(SupportsUpgradeDowngrade.class);
+ private final HoodieTable table = mock(HoodieTable.class);
+ private final HoodieWriteConfig config = HoodieWriteConfig.newBuilder()
+ .withPath("/table")
+ .withAutoUpgradeVersion(true)
+ .build();
+
+ TestContext(HoodieTableVersion version) throws Exception {
+ when(metaClient.getTableConfig()).thenReturn(tableConfig);
+ when(tableConfig.getTableVersion()).thenReturn(version);
+ when(tableConfig.isMetadataTableAvailable()).thenReturn(false);
+ when(metaClient.getStorage()).thenReturn(storage);
+ when(metaClient.getBasePath()).thenReturn(new StoragePath("/table"));
+ when(metaClient.getMetaPath()).thenReturn(new
StoragePath("/table/.hoodie"));
+
when(metaClient.getTableType()).thenReturn(HoodieTableType.COPY_ON_WRITE);
+ when(storage.exists(any(StoragePath.class))).thenReturn(false);
+ when(helper.getTable(config, engineContext)).thenReturn(table);
+ }
+
+ TrackingUpgradeDowngrade trackingUpgradeDowngrade() {
+ return new TrackingUpgradeDowngrade(metaClient, config, engineContext,
helper);
+ }
+ }
+
+ private static class TrackingUpgradeDowngrade extends UpgradeDowngrade {
+ private final List<String> hops = new ArrayList<>();
+ private RuntimeException failure;
+ private TableConfigChangeSet changeSet = new TableConfigChangeSet();
+
+ TrackingUpgradeDowngrade(HoodieTableMetaClient metaClient,
HoodieWriteConfig config,
+ HoodieEngineContext context,
SupportsUpgradeDowngrade helper) {
+ super(metaClient, config, context, helper);
+ }
+
+ @Override
+ protected TableConfigChangeSet upgrade(HoodieTableVersion fromVersion,
HoodieTableVersion toVersion, String instantTime) {
+ return recordHop(fromVersion, toVersion);
+ }
+
+ @Override
+ protected TableConfigChangeSet downgrade(HoodieTableVersion fromVersion,
HoodieTableVersion toVersion, String instantTime) {
+ return recordHop(fromVersion, toVersion);
+ }
+
+ private TableConfigChangeSet recordHop(HoodieTableVersion fromVersion,
HoodieTableVersion toVersion) {
+ hops.add(fromVersion.versionCode() + "->" + toVersion.versionCode());
+ if (failure != null) {
+ throw failure;
+ }
+ return changeSet;
+ }
+
+ private TableConfigChangeSet callRealUpgrade(HoodieTableVersion
fromVersion, HoodieTableVersion toVersion) {
+ return super.upgrade(fromVersion, toVersion, null);
+ }
+
+ private TableConfigChangeSet callRealDowngrade(HoodieTableVersion
fromVersion, HoodieTableVersion toVersion) {
+ return super.downgrade(fromVersion, toVersion, null);
+ }
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
index da6b4ac2c12b..6e74e60fdeae 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
@@ -19,6 +19,7 @@
package org.apache.hudi.table.upgrade;
import org.apache.hudi.DataSourceWriteOptions;
+import org.apache.hudi.avro.model.HoodieCompactionPlan;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.WriteClientTestUtils;
import org.apache.hudi.common.config.HoodieMetadataConfig;
@@ -117,6 +118,41 @@ public class TestUpgradeDowngrade extends
SparkClientFunctionalTestHarness {
assertTrue(originalMetaClient.getTableConfig().isMetadataTableAvailable());
}
+ @ParameterizedTest
+ @MethodSource("legacyFixtureUpgradePairs")
+ public void testLegacyUpgradeHandlersWithFixtureTables(
+ HoodieTableVersion fromVersion, HoodieTableVersion toVersion) throws
Exception {
+ HoodieTableMetaClient originalMetaClient = loadFixtureTable(fromVersion);
+ Dataset<Row> originalData = readTableData(originalMetaClient, "before
legacy fixture upgrade");
+ StoragePath metadataTablePath =
HoodieTableMetadata.getMetadataTableBasePath(originalMetaClient.getBasePath());
+ if (originalMetaClient.getStorage().exists(metadataTablePath)) {
+ originalMetaClient.getStorage().deleteDirectory(metadataTablePath);
+ }
+
+ StoragePath auxiliaryCompactionFile = null;
+ if (fromVersion == HoodieTableVersion.FIVE) {
+ HoodieInstant requestedCompaction =
originalMetaClient.getInstantGenerator().createNewInstant(
+ HoodieInstant.State.REQUESTED, HoodieTimeline.COMPACTION_ACTION,
"20250802170400000");
+ originalMetaClient.getActiveTimeline().saveToCompactionRequested(
+ requestedCompaction,
HoodieCompactionPlan.newBuilder().setVersion(1).build());
+ String instantFileName =
originalMetaClient.getInstantFileNameGenerator().getFileName(requestedCompaction);
+ auxiliaryCompactionFile = new
StoragePath(originalMetaClient.getMetaAuxiliaryPath(), instantFileName);
+ originalMetaClient.getStorage().create(auxiliaryCompactionFile,
false).close();
+
assertTrue(originalMetaClient.getStorage().exists(auxiliaryCompactionFile));
+ }
+
+ HoodieWriteConfig config = createWriteConfig(originalMetaClient, true);
+ new LegacyFixtureUpgradeDowngrade(
+ originalMetaClient, config, context(),
SparkUpgradeDowngradeHelper.getInstance()).run(toVersion, null);
+
+ HoodieTableMetaClient resultMetaClient =
HoodieTableMetaClient.reload(originalMetaClient);
+ assertEquals(toVersion,
resultMetaClient.getTableConfig().getTableVersion());
+ if (auxiliaryCompactionFile != null) {
+
assertFalse(resultMetaClient.getStorage().exists(auxiliaryCompactionFile));
+ }
+ validateDataConsistency(originalData, resultMetaClient, "after legacy
fixture upgrade");
+ }
+
@Disabled
@ParameterizedTest
@MethodSource("upgradeDowngradeVersionPairs")
@@ -692,6 +728,13 @@ public class TestUpgradeDowngrade extends
SparkClientFunctionalTestHarness {
);
}
+ private static Stream<Arguments> legacyFixtureUpgradePairs() {
+ return Stream.of(
+ Arguments.of(HoodieTableVersion.FOUR, HoodieTableVersion.FIVE),
+ Arguments.of(HoodieTableVersion.FIVE, HoodieTableVersion.SIX)
+ );
+ }
+
private static Stream<Arguments> versionsSixAndAbove() {
return Stream.of(
Arguments.of(HoodieTableVersion.SIX), // Hudi 0.14
@@ -1185,4 +1228,17 @@ public class TestUpgradeDowngrade extends
SparkClientFunctionalTestHarness {
validateDataConsistency(expectedDataWithNewRecord, metaClientV6,
"dataframe validation after v9->v6 downgrade for " + payloadType);
log.info("Completed payload upgrade/downgrade test for: {}", payloadType);
}
+
+ private static class LegacyFixtureUpgradeDowngrade extends UpgradeDowngrade {
+ LegacyFixtureUpgradeDowngrade(HoodieTableMetaClient metaClient,
HoodieWriteConfig config,
+
org.apache.hudi.common.engine.HoodieEngineContext context,
+ SupportsUpgradeDowngrade
upgradeDowngradeHelper) {
+ super(metaClient, config, context, upgradeDowngradeHelper);
+ }
+
+ @Override
+ public boolean needsUpgradeOrDowngrade(HoodieTableVersion toWriteVersion) {
+ return true;
+ }
+ }
}
