danny0405 commented on code in PR #19537: URL: https://github.com/apache/hudi/pull/19537#discussion_r3734709751
########## hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/upgrade/TestLegacyUpgradeDowngradeHandlers.java: ########## @@ -0,0 +1,352 @@ +/* + * 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.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.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 = org.mockito.Mockito.spy(new ZeroToOneUpgradeHandler()); + org.mockito.Mockito.doNothing().when(handler).recreateMarkers(anyString(), eq(table), eq(context), anyInt()); Review Comment: Addressed in d7887698fe76: added static imports for spy and doNothing and removed the fully qualified Mockito calls. The updated module passes Checkstyle. ########## hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/index/TestIndexingCatchupTask.java: ########## @@ -240,6 +249,121 @@ public void testNoHeartbeat() throws IOException { 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(); + currentCaughtupInstant = instant.requestedTime(); Review Comment: Addressed in d7887698fe76. AbstractIndexingCatchupTask now advances progress after every successfully processed completed action, and RunIndexActionExecutor reads that progress after Future.get() before building HoodieIndexCommitMetadata. The executor regression now runs the production write-stat catch-up task against a real completed 002 commit and asserts indexUptoInstant is 002. The full PR-focused suite passes all 46 tests. ########## hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/index/TestIndexActionExecutors.java: ########## @@ -0,0 +1,341 @@ +/* + * 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.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 IOException { + 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)); Review Comment: Addressed in d7887698fe76. The FILES initialization branch now owns the metadata writer with try-with-resources, and the tests verify close() after both successful initialization and a saveAsComplete timeline failure. The full PR-focused suite passes all 46 tests. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
