This is an automated email from the ASF dual-hosted git repository.
danny0405 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 a4373f75bad4 test(hudi-client): improve metadata table writer coverage
(#19363)
a4373f75bad4 is described below
commit a4373f75bad4e8175ab00b121706417b1cfaed7c
Author: Shuo Cheng <[email protected]>
AuthorDate: Fri Jul 24 13:55:54 2026 +0800
test(hudi-client): improve metadata table writer coverage (#19363)
* test(hudi-client): improve metadata table writer coverage
---
.../TestHoodieBackedTableMetadataWriter.java | 254 +++++++++++++++++
...ieBackedTableMetadataWriterTableVersionSix.java | 153 ++++++++++
.../metadata/TestHoodieMetadataWriteUtils.java | 45 +++
.../TestSecondaryIndexRecordGenerationUtils.java | 73 +++++
.../TestSparkHoodieBackedTableMetadataWriter.java | 183 ++++++++++++
.../functional/TestMetadataTableWithSparkSQL.scala | 311 +++++++++++++++++++++
6 files changed, 1019 insertions(+)
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriter.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriter.java
index 1a6d32a592e2..e3f4ae662182 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriter.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriter.java
@@ -18,40 +18,59 @@
package org.apache.hudi.metadata;
+import org.apache.hudi.avro.model.HoodieIndexPartitionInfo;
+import org.apache.hudi.avro.model.HoodieRestoreMetadata;
import org.apache.hudi.client.BaseHoodieWriteClient;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.HoodieTableServiceManagerConfig;
+import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
import org.apache.hudi.common.model.HoodieRecord;
+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.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.timeline.versioning.v2.ActiveTimelineV2;
+import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.util.Lazy;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieCleanConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieIndexException;
+import org.apache.hudi.metadata.index.Indexer;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.MockedStatic;
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
import java.util.stream.Stream;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
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.CALLS_REAL_METHODS;
@@ -252,6 +271,241 @@ class TestHoodieBackedTableMetadataWriter {
assertDoesNotThrow(() -> validateRollbackMethod.invoke(writer,
instantToRollback));
}
+ @Test
+ void exercisesNoOpAndUnsupportedBaseWriterPaths() throws Exception {
+ // Cover base no-op hooks and fail fast when an indexer is disabled.
+ HoodieBackedTableMetadataWriter<List<HoodieRecord>, List<?>> writer =
+ mock(HoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS);
+
+ assertDoesNotThrow(() -> writer.buildMetadataPartitions(
+ mock(HoodieEngineContext.class), Collections.emptyList(), "001"));
+ assertThrows(UnsupportedOperationException.class,
+ () -> writer.streamWriteToMetadataTable(null, "001"));
+ assertThrows(UnsupportedOperationException.class,
+ () ->
writer.secondaryWriteToMetadataTablePartitions(Collections.emptyList(), "001"));
+
+ HoodieWriteConfig metadataWriteConfig = mock(HoodieWriteConfig.class);
+ when(metadataWriteConfig.getBasePath()).thenReturn("/tmp/metadata");
+ writer.metadataWriteConfig = metadataWriteConfig;
+ setField(writer, "enabledIndexerMap", Collections.emptyMap());
+ HoodieIndexPartitionInfo disabledPartition =
HoodieIndexPartitionInfo.newBuilder()
+ .setVersion(1)
+
.setMetadataPartitionPath(MetadataPartitionType.COLUMN_STATS.getPartitionPath())
+ .setIndexUptoInstant("001")
+ .build();
+ assertThrows(HoodieIndexException.class,
+ () -> writer.buildMetadataPartitions(
+ mock(HoodieEngineContext.class),
Collections.singletonList(disabledPartition), "001"));
+ }
+
+ @Test
+ void fallsBackToConfiguredIndexersWhenTableConfigHasNoMetadataPartitions()
throws Exception {
+ // A fresh table must derive active indexers from writer configuration.
+ HoodieBackedTableMetadataWriter<List<HoodieRecord>, List<?>> writer =
+ mock(HoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS);
+ HoodieTableMetaClient dataMetaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+ when(dataMetaClient.getTableConfig()).thenReturn(tableConfig);
+ when(tableConfig.getMetadataPartitions()).thenReturn(new HashSet<>());
+
when(tableConfig.getMetadataPartitionsInflight()).thenReturn(Collections.emptySet());
+ writer.dataMetaClient = dataMetaClient;
+ Map<MetadataPartitionType, Indexer> enabledIndexers = new HashMap<>();
+ enabledIndexers.put(MetadataPartitionType.FILES, mock(Indexer.class));
+ setField(writer, "enabledIndexerMap", enabledIndexers);
+
+ assertDoesNotThrow(() -> writer.processAndCommit("001",
Collections::emptyList));
+
+ HoodieBackedTableMetadataWriter<List<HoodieRecord>, List<?>>
streamingWriter =
+ mock(HoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS);
+ HoodieEngineContext engineContext = mock(HoodieEngineContext.class);
+ HoodieData<HoodieRecord> emptyData = mock(HoodieData.class);
+ when(engineContext.<HoodieRecord>emptyHoodieData()).thenReturn(emptyData);
+ streamingWriter.dataMetaClient = dataMetaClient;
+ setField(streamingWriter, "engineContext", engineContext);
+ setField(streamingWriter, "enabledIndexerMap", Collections.emptyMap());
+ assertSame(emptyData, streamingWriter.streamWriteToMetadataPartitions(
+ mock(HoodieData.class), Collections.emptySet(), "001"));
+ }
+
+ @Test
+ void wrapsMetadataReaderAndFileSliceReadFailures() throws Exception {
+ // Reader setup and lazy file listing must preserve the public exception
contract.
+ HoodieBackedTableMetadataWriter<List<HoodieRecord>, List<?>> writer =
+ mock(HoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS);
+ writer.dataWriteConfig =
HoodieWriteConfig.newBuilder().withPath("/tmp/missing-table").build();
+ writer.dataMetaClient = mock(HoodieTableMetaClient.class);
+ Method maybeReinitializeReader =
+
HoodieBackedTableMetadataWriter.class.getDeclaredMethod("mayBeReinitMetadataReader");
+ maybeReinitializeReader.setAccessible(true);
+ InvocationTargetException readerFailure = assertThrows(
+ InvocationTargetException.class, () ->
maybeReinitializeReader.invoke(writer));
+ assertTrue(readerFailure.getCause() instanceof HoodieException);
+
+ HoodieBackedTableMetadata metadata = mock(HoodieBackedTableMetadata.class);
+ HoodieTableFileSystemView metadataView =
mock(HoodieTableFileSystemView.class);
+ HoodieTableMetaClient dataMetaClient = mock(HoodieTableMetaClient.class,
RETURNS_DEEP_STUBS);
+
when(dataMetaClient.getActiveTimeline().filterCompletedAndCompactionInstants().lastInstant())
+ .thenReturn(Option.empty());
+ when(metadata.getMetadataFileSystemView()).thenReturn(metadataView);
+ when(metadata.getAllPartitionPaths()).thenThrow(new IOException("listing
failed"));
+ writer.metadata = metadata;
+ writer.dataMetaClient = dataMetaClient;
+ setField(writer, "metadataView", metadataView);
+ Method getLazyMergedFileSlices =
+
HoodieBackedTableMetadataWriter.class.getDeclaredMethod("getLazyMergedFileSlices");
+ getLazyMergedFileSlices.setAccessible(true);
+ Lazy<?> lazyFileSlices = (Lazy<?>) getLazyMergedFileSlices.invoke(writer);
+ assertThrows(HoodieIOException.class, lazyFileSlices::get);
+ }
+
+ @Test
+ void detectsEmptyMetadataTimelineAndHandlesMissingMetadataTable(@TempDir
Path tempDir) throws Exception {
+ // Missing MDT state requires bootstrap without trusting stale table
config.
+ HoodieBackedTableMetadataWriter<List<HoodieRecord>, List<?>> writer =
+ mock(HoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS);
+ Method isBootstrapNeeded = HoodieBackedTableMetadataWriter.class
+ .getDeclaredMethod("isBootstrapNeeded", Option.class);
+ isBootstrapNeeded.setAccessible(true);
+ assertTrue((boolean) isBootstrapNeeded.invoke(writer, Option.empty()));
+
+ HoodieTableMetaClient dataMetaClient = mock(HoodieTableMetaClient.class);
+ HoodieTableConfig tableConfig = mock(HoodieTableConfig.class);
+ when(dataMetaClient.getTableConfig()).thenReturn(tableConfig);
+ when(tableConfig.isMetadataTableAvailable()).thenReturn(true);
+ writer.storageConf =
org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf();
+ writer.dataWriteConfig = HoodieWriteConfig.newBuilder()
+ .withPath(tempDir.resolve("data-table").toString())
+ .build();
+ writer.metadataWriteConfig = HoodieWriteConfig.newBuilder()
+ .withPath(tempDir.resolve("missing-metadata-table").toString())
+ .build();
+ Method metadataTableExists = HoodieBackedTableMetadataWriter.class
+ .getDeclaredMethod("metadataTableExists", HoodieTableMetaClient.class);
+ metadataTableExists.setAccessible(true);
+ assertFalse((boolean) metadataTableExists.invoke(writer, dataMetaClient));
+ }
+
+ @Test
+ void ignoresIOExceptionWhileRemovingPendingIndexInstant() throws Exception {
+ // A corrupt pending index plan must not block partition cleanup.
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline timeline = mock(HoodieActiveTimeline.class);
+ HoodieInstant pendingIndex = INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.REQUESTED, HoodieTimeline.INDEXING_ACTION, "001");
+ when(metaClient.getInstantGenerator()).thenReturn(INSTANT_GENERATOR);
+ when(metaClient.reloadActiveTimeline()).thenReturn(timeline);
+ when(metaClient.getActiveTimeline()).thenReturn(timeline);
+ when(timeline.filterPendingIndexTimeline()).thenReturn(timeline);
+ when(timeline.getInstantsAsStream()).thenReturn(Stream.of(pendingIndex));
+ when(timeline.readIndexPlan(pendingIndex)).thenThrow(new
IOException("cannot read plan"));
+ Method deletePendingIndexingInstant = HoodieBackedTableMetadataWriter.class
+ .getDeclaredMethod("deletePendingIndexingInstant",
HoodieTableMetaClient.class, String.class);
+ deletePendingIndexingInstant.setAccessible(true);
+
+ assertDoesNotThrow(() -> deletePendingIndexingInstant.invoke(null,
metaClient, "column_stats"));
+ }
+
+ @Test
+ void wrapsRestorePlanReadFailure() throws Exception {
+ // Restore-plan I/O failures must surface as HoodieIOException.
+ HoodieBackedTableMetadataWriter<List<HoodieRecord>, List<?>> writer =
+ mock(HoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS);
+ HoodieBackedTableMetadata metadata = mock(HoodieBackedTableMetadata.class);
+ HoodieTableFileSystemView metadataView =
mock(HoodieTableFileSystemView.class);
+ HoodieTableMetaClient metadataMetaClient =
mock(HoodieTableMetaClient.class);
+ HoodieTableMetaClient dataMetaClient = mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline timeline = mock(HoodieActiveTimeline.class);
+ when(metadata.getMetadataFileSystemView()).thenReturn(metadataView);
+ when(dataMetaClient.getInstantGenerator()).thenReturn(INSTANT_GENERATOR);
+ when(dataMetaClient.getActiveTimeline()).thenReturn(timeline);
+ when(timeline.readRestorePlan(any())).thenThrow(new IOException("cannot
read restore plan"));
+ writer.metadata = metadata;
+ writer.metadataMetaClient = metadataMetaClient;
+ writer.dataMetaClient = dataMetaClient;
+
+ assertThrows(HoodieIOException.class,
+ () -> writer.update(mock(HoodieRestoreMetadata.class), "001"));
+ }
+
+ @Test
+ void rejectsPendingMetadataCompactionAndWrapsCloseFailures() {
+ // Pending compaction blocks scheduling, while close errors remain visible.
+ HoodieBackedTableMetadataWriter<List<HoodieRecord>, List<?>> writer =
+ mock(HoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS);
+ HoodieWriteConfig metadataWriteConfig = mock(HoodieWriteConfig.class);
+ when(metadataWriteConfig.isLogCompactionEnabled()).thenReturn(true);
+ writer.metadataWriteConfig = metadataWriteConfig;
+ HoodieTableMetaClient metadataMetaClient =
mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline metadataTimeline = mock(HoodieActiveTimeline.class,
RETURNS_DEEP_STUBS);
+ HoodieInstant pendingCompaction = INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.REQUESTED, HoodieTimeline.COMPACTION_ACTION,
"001");
+ when(metadataMetaClient.getActiveTimeline()).thenReturn(metadataTimeline);
+
when(metadataTimeline.filterPendingLogCompactionTimeline().firstInstant()).thenReturn(Option.empty());
+
when(metadataTimeline.filterPendingCompactionTimeline().firstInstant()).thenReturn(Option.of(pendingCompaction));
+ writer.metadataMetaClient = metadataMetaClient;
+
+ assertThrows(HoodieException.class, () -> {
+ doThrow(new HoodieException("close failed")).when(writer).close();
+ writer.closeInternal();
+ });
+ assertFalse(writer.validateCompactionScheduling(Option.empty(), "002"));
+ }
+
+ @Test
+ void compactIfNecessaryHandlesSkipDelegationAndFailures() {
+ // Exercise skip, delegation, and failure propagation for both compaction
types.
+ Properties tableServiceManagerProperties = new Properties();
+ tableServiceManagerProperties.put(
+ HoodieTableServiceManagerConfig.TABLE_SERVICE_MANAGER_ENABLED.key(),
"true");
+ tableServiceManagerProperties.put(
+ HoodieTableServiceManagerConfig.TABLE_SERVICE_MANAGER_ACTIONS.key(),
"compaction,logcompaction");
+ HoodieTableServiceManagerConfig tableServiceManagerConfig =
+
HoodieTableServiceManagerConfig.newBuilder().fromProperties(tableServiceManagerProperties).build();
+ HoodieWriteConfig metadataWriteConfig = mock(HoodieWriteConfig.class);
+
when(metadataWriteConfig.getTableServiceManagerConfig()).thenReturn(tableServiceManagerConfig);
+ when(metadataWriteConfig.isLogCompactionEnabled()).thenReturn(true);
+
+ HoodieTableMetaClient dataMetaClient = mock(HoodieTableMetaClient.class,
RETURNS_DEEP_STUBS);
+ when(dataMetaClient.reloadActiveTimeline().filterInflightsAndRequested()
+ .filter(any()).firstInstant()).thenReturn(Option.empty());
+ HoodieTableMetaClient metadataMetaClient =
mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline metadataTimeline = mock(HoodieActiveTimeline.class);
+ HoodieTimeline completedTimeline = mock(HoodieTimeline.class);
+ when(metadataMetaClient.getActiveTimeline()).thenReturn(metadataTimeline);
+
when(metadataTimeline.filterCompletedInstants()).thenReturn(completedTimeline);
+ when(completedTimeline.containsInstant(any(String.class)))
+ .thenAnswer(invocation -> "100".equals(invocation.getArgument(0)));
+
+ HoodieBackedTableMetadataWriter<List<HoodieRecord>, List<?>> writer =
+ mock(HoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS);
+ writer.dataMetaClient = dataMetaClient;
+ writer.metadataMetaClient = metadataMetaClient;
+ writer.metadataWriteConfig = metadataWriteConfig;
+ writer.metrics = Option.empty();
+
+ BaseHoodieWriteClient writeClient = mock(BaseHoodieWriteClient.class);
+ when(writeClient.createNewInstantTime(false)).thenReturn("100", "200",
"300", "400");
+ when(writeClient.scheduleCompactionAtInstant("200",
Option.empty())).thenReturn(true);
+ when(writeClient.scheduleCompactionAtInstant("300", Option.empty()))
+ .thenThrow(new HoodieException("compaction failed"));
+ when(writeClient.scheduleCompactionAtInstant("400",
Option.empty())).thenReturn(false);
+ when(writeClient.scheduleLogCompaction(Option.empty()))
+ .thenReturn(Option.of("201"))
+ .thenThrow(new HoodieException("log compaction failed"));
+
+ writer.compactIfNecessary(writeClient, Option.empty());
+ writer.compactIfNecessary(writeClient, Option.empty());
+ assertThrows(HoodieException.class, () ->
writer.compactIfNecessary(writeClient, Option.empty()));
+ assertThrows(HoodieException.class, () ->
writer.compactIfNecessary(writeClient, Option.empty()));
+ }
+
+ private static void setField(Object target, String name, Object value)
throws Exception {
+ // Exercise private failure paths without changing production visibility.
+ java.lang.reflect.Field field =
HoodieBackedTableMetadataWriter.class.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+
@SuppressWarnings("deprecation")
private HoodieActiveTimeline createMockTimeline(List<HoodieInstant>
instants) {
ActiveTimelineV2 timeline = new ActiveTimelineV2();
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriterTableVersionSix.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriterTableVersionSix.java
index 804acc5f2ef3..184003552ca3 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriterTableVersionSix.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieBackedTableMetadataWriterTableVersionSix.java
@@ -18,6 +18,10 @@
package org.apache.hudi.metadata;
+import org.apache.hudi.client.BaseHoodieWriteClient;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.HoodieTableServiceManagerConfig;
+import org.apache.hudi.common.model.ActionType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
@@ -26,19 +30,28 @@ import
org.apache.hudi.common.table.timeline.InstantGenerator;
import org.apache.hudi.common.table.timeline.versioning.v1.ActiveTimelineV1;
import org.apache.hudi.common.table.timeline.versioning.v1.InstantGeneratorV1;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieMetadataException;
import org.junit.jupiter.api.Test;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Properties;
import java.util.Set;
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.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
@@ -225,6 +238,146 @@ class TestHoodieBackedTableMetadataWriterTableVersionSix {
assertFalse(result, "Should block initialization when rollbacks are
completed, not pending");
}
+ @Test
+ void testGenerateUniqueInstantTimePreservesIndexingInstant() throws
Exception {
+ // An indexing instant is already globally unique and must be reused.
+ String indexingInstant = "20250101120000000";
+ HoodieTableMetaClient dataMetaClient = mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline timeline =
createMockTimeline(Collections.singletonList(
+ INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.REQUESTED, HoodieTimeline.INDEXING_ACTION,
indexingInstant)));
+ when(dataMetaClient.getActiveTimeline()).thenReturn(timeline);
+ HoodieBackedTableMetadataWriterTableVersionSix<?, ?> writer =
createMockWriter(dataMetaClient);
+
+
assertTrue(indexingInstant.equals(writer.generateUniqueInstantTime(indexingInstant)));
+ }
+
+ @Test
+ void testValidateCompactionSchedulingRejectsPendingMetadataTableService()
throws Exception {
+ // Do not schedule compaction while another metadata table service is
pending.
+ HoodieTableMetaClient dataMetaClient = mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline dataTimeline =
createMockTimeline(Collections.emptyList());
+ when(dataMetaClient.reloadActiveTimeline()).thenReturn(dataTimeline);
+ HoodieBackedTableMetadataWriterTableVersionSix<?, ?> writer =
createMockWriter(dataMetaClient);
+
+ HoodieTableMetaClient metadataMetaClient =
mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline metadataTimeline = mock(HoodieActiveTimeline.class,
RETURNS_DEEP_STUBS);
+ HoodieInstant pendingCompaction = INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.REQUESTED, HoodieTimeline.COMPACTION_ACTION,
"20250101120000001");
+ when(metadataMetaClient.getActiveTimeline()).thenReturn(metadataTimeline);
+
when(metadataTimeline.filterPendingLogCompactionTimeline().firstInstant()).thenReturn(Option.empty());
+
when(metadataTimeline.filterPendingCompactionTimeline().firstInstant()).thenReturn(Option.of(pendingCompaction));
+ writer.metadataMetaClient = metadataMetaClient;
+
+ assertFalse(writer.validateCompactionScheduling(Option.empty(),
"20250101120000002"));
+ }
+
+ @Test
+ void testValidateCompactionSchedulingRejectsExcessiveDeltaCommits() throws
Exception {
+ // Protect pending data commits from unbounded metadata delta commits.
+ HoodieTableMetaClient dataMetaClient = mock(HoodieTableMetaClient.class);
+ HoodieInstant pendingCommit = INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.REQUESTED, HoodieTimeline.COMMIT_ACTION,
"20250101120000000");
+ HoodieActiveTimeline dataTimeline =
createMockTimeline(Collections.singletonList(pendingCommit));
+ when(dataMetaClient.reloadActiveTimeline()).thenReturn(dataTimeline);
+ HoodieBackedTableMetadataWriterTableVersionSix<?, ?> writer =
createMockWriter(dataMetaClient);
+
+ HoodieTableMetaClient metadataMetaClient =
mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline metadataTimeline = mock(HoodieActiveTimeline.class,
RETURNS_DEEP_STUBS);
+
when(metadataMetaClient.reloadActiveTimeline()).thenReturn(metadataTimeline);
+
when(metadataTimeline.filterCompletedInstants().filter(any()).lastInstant()).thenReturn(Option.empty());
+
when(metadataTimeline.getDeltaCommitTimeline().countInstants()).thenReturn(2);
+ writer.metadataMetaClient = metadataMetaClient;
+ writer.dataWriteConfig = HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/table")
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+ .withMaxNumDeltacommitsWhenPending(1)
+ .build())
+ .build();
+
+ assertThrows(HoodieMetadataException.class,
+ () -> writer.validateCompactionScheduling(Option.empty(),
"20250101120000001"));
+ }
+
+ @Test
+ void testCompactIfNecessaryCoversExistingDelegatedAndLogCompactionPaths() {
+ // Exercise completed, delegated, and log-compaction fallback paths.
+ Properties tableServiceManagerProperties = new Properties();
+ tableServiceManagerProperties.put(
+ HoodieTableServiceManagerConfig.TABLE_SERVICE_MANAGER_ENABLED.key(),
"true");
+ tableServiceManagerProperties.put(
+ HoodieTableServiceManagerConfig.TABLE_SERVICE_MANAGER_ACTIONS.key(),
ActionType.compaction.name());
+ HoodieTableServiceManagerConfig tableServiceManagerConfig =
+ HoodieTableServiceManagerConfig.newBuilder()
+ .fromProperties(tableServiceManagerProperties)
+ .build();
+ HoodieWriteConfig metadataWriteConfig = mock(HoodieWriteConfig.class);
+
when(metadataWriteConfig.getTableServiceManagerConfig()).thenReturn(tableServiceManagerConfig);
+ when(metadataWriteConfig.isLogCompactionEnabled()).thenReturn(true);
+
+ HoodieTableMetaClient metadataMetaClient =
mock(HoodieTableMetaClient.class);
+ HoodieActiveTimeline timeline = mock(HoodieActiveTimeline.class,
RETURNS_DEEP_STUBS);
+ when(metadataMetaClient.getActiveTimeline()).thenReturn(timeline);
+
when(timeline.filterCompletedInstants().containsInstant("100001")).thenReturn(true);
+
when(timeline.filterCompletedInstants().containsInstant("200001")).thenReturn(false);
+
when(timeline.filterCompletedInstants().containsInstant("300001")).thenReturn(false);
+
when(timeline.filterCompletedInstants().containsInstant("300005")).thenReturn(false);
+
when(timeline.filterCompletedInstants().containsInstant("400001")).thenReturn(false);
+
when(timeline.filterCompletedInstants().containsInstant("400005")).thenReturn(false);
+
+ HoodieBackedTableMetadataWriterTableVersionSix<?, ?> writer =
+ mock(HoodieBackedTableMetadataWriterTableVersionSix.class,
CALLS_REAL_METHODS);
+ writer.metadataMetaClient = metadataMetaClient;
+ writer.metadataWriteConfig = metadataWriteConfig;
+ BaseHoodieWriteClient writeClient = mock(BaseHoodieWriteClient.class);
+ when(writeClient.scheduleCompactionAtInstant("200001",
Option.empty())).thenReturn(true);
+ when(writeClient.scheduleCompactionAtInstant("300001",
Option.empty())).thenReturn(false);
+ when(writeClient.scheduleLogCompactionAtInstant("300005",
Option.empty())).thenReturn(true);
+
+ // Version 6 derives compaction and log-compaction instants with fixed
suffixes.
+ writer.compactIfNecessary(writeClient, Option.of("100"));
+ writer.compactIfNecessary(writeClient, Option.of("200"));
+ writer.compactIfNecessary(writeClient, Option.of("300"));
+
+ Properties allTableServicesProperties = new Properties();
+ allTableServicesProperties.put(
+ HoodieTableServiceManagerConfig.TABLE_SERVICE_MANAGER_ENABLED.key(),
"true");
+ allTableServicesProperties.put(
+ HoodieTableServiceManagerConfig.TABLE_SERVICE_MANAGER_ACTIONS.key(),
"compaction,logcompaction");
+ when(metadataWriteConfig.getTableServiceManagerConfig()).thenReturn(
+ HoodieTableServiceManagerConfig.newBuilder()
+ .fromProperties(allTableServicesProperties)
+ .build());
+ when(writeClient.scheduleCompactionAtInstant("400001",
Option.empty())).thenReturn(false);
+ when(writeClient.scheduleLogCompactionAtInstant("400005",
Option.empty())).thenReturn(true);
+ writer.compactIfNecessary(writeClient, Option.of("400"));
+
+ verify(writeClient).scheduleCompactionAtInstant("200001", Option.empty());
+ verify(writeClient).scheduleLogCompactionAtInstant("300005",
Option.empty());
+ verify(writeClient).logCompact("300005", true);
+ }
+
+ @Test
+ void testValidateRollbackRejectsCommitBeforeLatestCompaction() throws
Exception {
+ // Version 6 cannot roll back beyond the latest compaction boundary.
+ HoodieBackedTableMetadataWriterTableVersionSix<?, ?> writer =
+ mock(HoodieBackedTableMetadataWriterTableVersionSix.class,
CALLS_REAL_METHODS);
+ HoodieInstant compactionInstant = INSTANT_GENERATOR.createNewInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "200",
"201");
+ HoodieTimeline deltaCommits = mock(HoodieTimeline.class);
+ when(deltaCommits.countInstants()).thenReturn(2);
+ when(deltaCommits.getInstants()).thenReturn(Collections.emptyList());
+ Method validateRollback =
HoodieBackedTableMetadataWriterTableVersionSix.class
+ .getDeclaredMethod(
+ "validateRollbackVersionSix", String.class, HoodieInstant.class,
HoodieTimeline.class);
+ validateRollback.setAccessible(true);
+
+ InvocationTargetException exception = assertThrows(
+ InvocationTargetException.class,
+ () -> validateRollback.invoke(writer, "100", compactionInstant,
deltaCommits));
+ assertTrue(exception.getCause() instanceof HoodieMetadataException);
+ }
+
private HoodieBackedTableMetadataWriterTableVersionSix<?, ?>
createMockWriter(HoodieTableMetaClient dataMetaClient) throws Exception {
// Use CALLS_REAL_METHODS so that shouldInitializeFromFilesystem executes
the real logic
HoodieBackedTableMetadataWriterTableVersionSix<?, ?> writer =
mock(HoodieBackedTableMetadataWriterTableVersionSix.class, CALLS_REAL_METHODS);
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataWriteUtils.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataWriteUtils.java
index 296716fcaaa3..a68bd289119a 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataWriteUtils.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataWriteUtils.java
@@ -24,6 +24,7 @@ import
org.apache.hudi.client.transaction.lock.ZookeeperBasedImplicitBasePathLoc
import org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider;
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.config.metrics.HoodieMetricsConfig;
import org.apache.hudi.common.engine.HoodieLocalEngineContext;
import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.fs.FSUtils;
@@ -46,15 +47,19 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.core.io.storage.HoodieIOFactory;
import org.apache.hudi.core.transaction.lock.InProcessLockProvider;
import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieMetadataException;
+import org.apache.hudi.metrics.MetricsReporterType;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.nio.file.Path;
+import java.util.Collections;
import java.util.Map;
import java.util.Properties;
@@ -124,6 +129,46 @@ public class TestHoodieMetadataWriteUtils {
assertEquals(hfileBloomFilterEnabled,
metadataWriteConfig.hfileBloomFilterEnabled());
}
+ @ParameterizedTest
+ @EnumSource(value = MetricsReporterType.class, names = {
+ "GRAPHITE", "JMX", "PROMETHEUS_PUSHGATEWAY", "M3", "PROMETHEUS"
+ })
+ void
testCreateMetadataWriteConfigPropagatesSupportedMetricsReporter(MetricsReporterType
reporterType) {
+ // Metadata writes must retain every reporter supported by the writer path.
+ HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/base_path/")
+ .withProps(Collections.singletonMap(
+ "hoodie.metrics.graphite.metric.prefix", "metadata-test"))
+ .withMetricsConfig(HoodieMetricsConfig.newBuilder()
+ .on(true)
+ .withReporterType(reporterType.name())
+ .build())
+ .build();
+
+ HoodieWriteConfig metadataWriteConfig =
HoodieMetadataWriteUtils.createMetadataWriteConfig(
+ writeConfig, HoodieFailedWritesCleaningPolicy.EAGER,
HoodieTableVersion.EIGHT);
+
+ assertTrue(metadataWriteConfig.isMetricsOn());
+ assertEquals(reporterType, metadataWriteConfig.getMetricsReporterType());
+ }
+
+ @Test
+ void testCreateMetadataWriteConfigRejectsUnsupportedMetricsReporter() {
+ // Reject unsupported reporters before the metadata writer starts.
+ HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()
+ .withPath("/tmp/base_path/")
+ .withMetricsConfig(HoodieMetricsConfig.newBuilder()
+ .on(true)
+ .withReporterType(MetricsReporterType.SLF4J.name())
+ .build())
+ .build();
+
+ HoodieMetadataException exception =
assertThrows(HoodieMetadataException.class,
+ () -> HoodieMetadataWriteUtils.createMetadataWriteConfig(
+ writeConfig, HoodieFailedWritesCleaningPolicy.EAGER,
HoodieTableVersion.EIGHT));
+ assertTrue(exception.getMessage().contains("Unsupported Metrics Reporter
type SLF4J"));
+ }
+
@Test
public void testCreateMetadataWriteConfigForCleaner() {
HoodieWriteConfig writeConfig1 = HoodieWriteConfig.newBuilder()
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestSecondaryIndexRecordGenerationUtils.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestSecondaryIndexRecordGenerationUtils.java
new file mode 100644
index 000000000000..3db50783cc6c
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/metadata/TestSecondaryIndexRecordGenerationUtils.java
@@ -0,0 +1,73 @@
+/*
+ * 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.metadata;
+
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+
+/**
+ * Tests validation and error propagation before secondary-index record
generation.
+ */
+class TestSecondaryIndexRecordGenerationUtils {
+
+ @Test
+ void rejectsLogFileInsertsBeforeReadingFileSlices() {
+ // Log-file inserts cannot be reconstructed without a base-file slice.
+ HoodieWriteStat writeStat = new HoodieWriteStat();
+ writeStat.setPartitionPath("p1");
+ writeStat.setPath("p1/.fileid-1_014.log.1_1-0-1");
+ writeStat.setNumInserts(1);
+ HoodieWriteConfig writeConfig =
HoodieWriteConfig.newBuilder().withPath("/tmp/table").build();
+
+ assertThrows(HoodieIOException.class,
+ () ->
SecondaryIndexRecordGenerationUtils.convertWriteStatsToSecondaryIndexRecords(
+ Collections.singletonList(writeStat), "001", null, null, null,
null, writeConfig));
+ }
+
+ @Test
+ void wrapsTableSchemaResolutionFailure() {
+ // Wrap schema lookup failures in the metadata utility's exception type.
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ HoodieWriteConfig writeConfig =
HoodieWriteConfig.newBuilder().withPath("/tmp/table").build();
+
+ try (MockedStatic<HoodieTableMetadataUtil> metadataUtil =
+ mockStatic(HoodieTableMetadataUtil.class, CALLS_REAL_METHODS)) {
+ metadataUtil.when(() ->
HoodieTableMetadataUtil.tryResolveSchemaForTable(metaClient))
+ .thenThrow(new IllegalStateException("no schema"));
+
+ assertThrows(HoodieException.class,
+ () ->
SecondaryIndexRecordGenerationUtils.convertWriteStatsToSecondaryIndexRecords(
+ Collections.emptyList(), "001", null, null, metaClient, null,
writeConfig));
+ }
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestSparkHoodieBackedTableMetadataWriter.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestSparkHoodieBackedTableMetadataWriter.java
new file mode 100644
index 000000000000..370d71512536
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestSparkHoodieBackedTableMetadataWriter.java
@@ -0,0 +1,183 @@
+/*
+ * 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.metadata;
+
+import org.apache.hudi.client.BaseHoodieWriteClient;
+import org.apache.hudi.client.HoodieWriteResult;
+import org.apache.hudi.client.SparkRDDMetadataWriteClient;
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.engine.EngineType;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieNotSupportedException;
+import org.apache.hudi.index.HoodieSparkIndexClient;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+
+import java.util.Collections;
+
+import static
org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
+import static
org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+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.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests Spark-specific behavior shared by current and table-version-six
metadata writers.
+ */
+class TestSparkHoodieBackedTableMetadataWriter {
+
+ @Test
+ void exposesSparkEngineAndRejectsV6StreamingConversion() {
+ // Both implementations use Spark, but version 6 has no streaming
conversion.
+ SparkHoodieBackedTableMetadataWriter currentWriter =
+ mock(SparkHoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS);
+ SparkHoodieBackedTableMetadataWriterTableVersionSix versionSixWriter =
+ mock(SparkHoodieBackedTableMetadataWriterTableVersionSix.class,
CALLS_REAL_METHODS);
+
+ assertEquals(EngineType.SPARK, currentWriter.getEngineType());
+ assertEquals(EngineType.SPARK, versionSixWriter.getEngineType());
+ assertThrows(HoodieNotSupportedException.class,
+ () -> versionSixWriter.convertEngineSpecificDataToHoodieData(null));
+ }
+
+ @Test
+ void versionSixFileGroupUpsertCommitsWriteStatuses() {
+ // Version 6 file-group writes use a prepped-record upsert and delta
commit.
+ SparkHoodieBackedTableMetadataWriterTableVersionSix writer =
+ mock(SparkHoodieBackedTableMetadataWriterTableVersionSix.class,
CALLS_REAL_METHODS);
+ BaseHoodieWriteClient<?, JavaRDD<HoodieRecord>, ?, JavaRDD<WriteStatus>>
writeClient =
+ mock(BaseHoodieWriteClient.class);
+ JavaRDD<HoodieRecord> records = mock(JavaRDD.class);
+ JavaRDD<WriteStatus> writeStatuses = mock(JavaRDD.class);
+ when(writeClient.upsertPreppedRecords(records,
"001")).thenReturn(writeStatuses);
+
+ writer.upsertAndCommit(writeClient, "001", records,
Collections.emptyList());
+
+ verify(writeClient).commit(
+ "001", writeStatuses, Option.empty(), DELTA_COMMIT_ACTION,
Collections.emptyMap());
+ }
+
+ @Test
+ void currentFileGroupUpsertCommitsFirstUpsertStatuses() {
+ // Current writer uses first-upsert semantics when file groups are
supplied.
+ SparkHoodieBackedTableMetadataWriter writer =
+ mock(SparkHoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS);
+ SparkRDDMetadataWriteClient writeClient =
mock(SparkRDDMetadataWriteClient.class);
+ JavaRDD<HoodieRecord> records = mock(JavaRDD.class);
+ JavaRDD<WriteStatus> writeStatuses = mock(JavaRDD.class);
+ java.util.List<HoodieFileGroupId> fileGroups = Collections.emptyList();
+ // The writer resolves its internal client even when one is passed in.
+ doReturn(writeClient).when(writer).getWriteClient();
+ when(writeClient.firstUpsertPreppedRecords(records, "003",
fileGroups)).thenReturn(writeStatuses);
+
+ writer.upsertAndCommit(writeClient, "003", records, fileGroups);
+
+ verify(writeClient).commit(
+ "003", writeStatuses, Option.empty(), DELTA_COMMIT_ACTION,
Collections.emptyMap());
+ }
+
+ @Test
+ void currentUpsertCoalescesRecordPreparationInput() {
+ // Record preparation honors its configured parallelism before writing.
+ SparkHoodieBackedTableMetadataWriter writer =
+ mock(SparkHoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS);
+ HoodieMetadataConfig metadataConfig = mock(HoodieMetadataConfig.class);
+ HoodieWriteConfig dataWriteConfig = mock(HoodieWriteConfig.class);
+ when(dataWriteConfig.getMetadataConfig()).thenReturn(metadataConfig);
+ when(metadataConfig.getRecordPreparationParallelism()).thenReturn(1);
+ writer.dataWriteConfig = dataWriteConfig;
+
+ BaseHoodieWriteClient<?, JavaRDD<HoodieRecord>, ?, JavaRDD<WriteStatus>>
writeClient =
+ mock(BaseHoodieWriteClient.class);
+ JavaRDD<HoodieRecord> records = mock(JavaRDD.class);
+ JavaRDD<HoodieRecord> coalescedRecords = mock(JavaRDD.class);
+ JavaRDD<WriteStatus> writeStatuses = mock(JavaRDD.class);
+ when(records.getNumPartitions()).thenReturn(2);
+ when(records.coalesce(1)).thenReturn(coalescedRecords);
+ when(writeClient.upsertPreppedRecords(coalescedRecords,
"004")).thenReturn(writeStatuses);
+
+ writer.upsertAndCommit(writeClient, "004", records);
+
+ verify(writeClient).commit(
+ "004", writeStatuses, Option.empty(), DELTA_COMMIT_ACTION,
Collections.emptyMap());
+ }
+
+ @Test
+ void bothSparkWritersUpdateColumnStatsDefinition() {
+ // Both writer versions delegate column-stat definitions to the Spark
index client.
+ SparkHoodieBackedTableMetadataWriter currentWriter =
+ mock(SparkHoodieBackedTableMetadataWriter.class, CALLS_REAL_METHODS);
+ SparkHoodieBackedTableMetadataWriterTableVersionSix versionSixWriter =
+ mock(SparkHoodieBackedTableMetadataWriterTableVersionSix.class,
CALLS_REAL_METHODS);
+ java.util.List<String> columns = Collections.singletonList("rider");
+
+ // Capture index clients created internally by both writer versions.
+ try (MockedConstruction<HoodieSparkIndexClient> construction =
+ mockConstruction(HoodieSparkIndexClient.class)) {
+ currentWriter.updateColumnsToIndexWithColStats(columns);
+ versionSixWriter.updateColumnsToIndexWithColStats(columns);
+
+ assertEquals(2, construction.constructed().size());
+ verify(construction.constructed().get(0))
+ .createOrUpdateColumnStatsIndexDefinition(null, columns);
+ verify(construction.constructed().get(1))
+ .createOrUpdateColumnStatsIndexDefinition(null, columns);
+ }
+ }
+
+ @Test
+ void versionSixDeletePartitionsCommitsReplaceCommit() {
+ // Deleting an MDT partition is committed as a replace commit.
+ SparkHoodieBackedTableMetadataWriterTableVersionSix writer =
+ mock(SparkHoodieBackedTableMetadataWriterTableVersionSix.class,
CALLS_REAL_METHODS);
+ SparkRDDWriteClient writeClient = mock(SparkRDDWriteClient.class);
+ HoodieWriteResult writeResult = mock(HoodieWriteResult.class);
+ JavaRDD<WriteStatus> writeStatuses = mock(JavaRDD.class);
+ HoodieTableMetaClient metadataMetaClient =
mock(HoodieTableMetaClient.class);
+ writer.metadataMetaClient = metadataMetaClient;
+ doReturn(writeClient).when(writer).getWriteClient();
+ when(writeResult.getWriteStatuses()).thenReturn(writeStatuses);
+
when(writeResult.getPartitionToReplaceFileIds()).thenReturn(Collections.emptyMap());
+ when(writeClient.deletePartitions(
+
Collections.singletonList(MetadataPartitionType.RECORD_INDEX.getPartitionPath()),
"002"))
+ .thenReturn(writeResult);
+
+ writer.deletePartitions("002",
Collections.singletonList(MetadataPartitionType.RECORD_INDEX));
+
+ verify(writeClient).startCommitForMetadataTable(eq(metadataMetaClient),
eq("002"), any());
+ verify(writeClient).commit(
+ "002", writeStatuses, Option.empty(), REPLACE_COMMIT_ACTION,
Collections.emptyMap());
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMetadataTableWithSparkSQL.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMetadataTableWithSparkSQL.scala
new file mode 100644
index 000000000000..3b063b2932e4
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestMetadataTableWithSparkSQL.scala
@@ -0,0 +1,311 @@
+/*
+ * 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.functional
+
+import org.apache.hudi.DataSourceWriteOptions._
+import org.apache.hudi.client.SparkRDDWriteClient
+import org.apache.hudi.client.common.HoodieSparkEngineContext
+import org.apache.hudi.common.config.{HoodieMetadataConfig, TypedProperties}
+import org.apache.hudi.common.model.HoodieTableType
+import org.apache.hudi.common.table.{HoodieTableMetaClient,
HoodieTableVersion, TableSchemaResolver}
+import org.apache.hudi.common.testutils.HoodieTestUtils
+import org.apache.hudi.config.{HoodieLockConfig, HoodieWriteConfig}
+import org.apache.hudi.core.transaction.lock.InProcessLockProvider
+import
org.apache.hudi.metadata.HoodieMetadataPayload.SECONDARY_INDEX_RECORD_KEY_SEPARATOR
+import org.apache.hudi.metadata.MetadataPartitionType
+import
org.apache.hudi.testutils.SparkClientFunctionalTestHarness.getSparkSqlConf
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarnessScala
+
+import org.apache.spark.SparkConf
+import org.junit.jupiter.api.{BeforeEach, Tag, Test}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
+
+import scala.collection.JavaConverters.mapAsJavaMapConverter
+
+/**
+ * Exercises {@code SparkHoodieBackedTableMetadataWriter} and its
table-version-six
+ * implementation through real Spark writes.
+ */
+@Tag("functional-c")
+class TestMetadataTableWithSparkSQL extends
SparkClientFunctionalTestHarnessScala {
+
+ override def conf: SparkConf = conf(getSparkSqlConf)
+
+ @BeforeEach
+ override def runBeforeEach(): Unit = {
+ super.runBeforeEach()
+ spark.sql(s"set ${HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key()} =
${classOf[InProcessLockProvider].getName}")
+ }
+
+ @Test
+ def testAllMetadataIndexesAcrossUpsertAndRollback(): Unit = {
+ // Validate the full MDT lifecycle against a current-version COW table.
+ val tableName = "metadata_writer_all_indexes"
+ val tablePath = s"$basePath/$tableName"
+ val writeOptions = metadataWriteOptions(tableName, streamingWrites = true)
+
+ // Bootstrap all supported MDT partitions through real writes.
+ createTable(tableName, tablePath, tableVersion = None)
+ spark.sql(
+ s"""insert into $tableName values
+ | (1, 'row1', 'alpha', 'p1'),
+ | (2, 'row2', 'beta', 'p1'),
+ | (3, 'row3', 'gamma', 'p2')
+ |""".stripMargin)
+ spark.sql(s"create index idx_rider on $tableName (rider)")
+
+ var metaClient = createMetaClient(tablePath)
+ assertMetadataPartitions(metaClient, includePartitionStats = true,
includeSecondaryIndex = true)
+ assertCommonMetadataRecords(tablePath, expectedRecordIndexCount = 3,
includePartitionStats = true)
+ // Metadata payload type 7 stores secondary-index mappings.
+ checkAnswer(s"select key from hudi_metadata('$tablePath') where type=7")(
+ Seq(s"alpha${SECONDARY_INDEX_RECORD_KEY_SEPARATOR}row1"),
+ Seq(s"beta${SECONDARY_INDEX_RECORD_KEY_SEPARATOR}row2"),
+ Seq(s"gamma${SECONDARY_INDEX_RECORD_KEY_SEPARATOR}row3")
+ )
+
+ // Rebuild the secondary index and verify its records are removed and
restored.
+ spark.sql(s"drop index idx_rider on $tableName")
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+
assertFalse(metaClient.getTableConfig.getMetadataPartitions.contains("secondary_index_idx_rider"))
+ assertEquals(0L, spark.sql(s"select key from hudi_metadata('$tablePath')
where type=7").count())
+
+ spark.sql(s"create index idx_rider on $tableName (rider)")
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+ assertMetadataPartitions(metaClient, includePartitionStats = true,
includeSecondaryIndex = true)
+ checkAnswer(s"select key from hudi_metadata('$tablePath') where type=7")(
+ Seq(s"alpha${SECONDARY_INDEX_RECORD_KEY_SEPARATOR}row1"),
+ Seq(s"beta${SECONDARY_INDEX_RECORD_KEY_SEPARATOR}row2"),
+ Seq(s"gamma${SECONDARY_INDEX_RECORD_KEY_SEPARATOR}row3")
+ )
+
+ // Upsert changes both data and the secondary-index key.
+ spark.sql(s"update $tableName set rider = 'delta', ts = 4 where id =
'row1'")
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+ val upsertInstant = metaClient.getActiveTimeline.getCommitsTimeline
+ .filterCompletedInstants.lastInstant().get().requestedTime()
+
+ checkAnswer(s"select id, rider, part from $tableName order by id")(
+ Seq("row1", "delta", "p1"),
+ Seq("row2", "beta", "p1"),
+ Seq("row3", "gamma", "p2")
+ )
+ checkAnswer(s"select key from hudi_metadata('$tablePath') where type=7")(
+ Seq(s"delta${SECONDARY_INDEX_RECORD_KEY_SEPARATOR}row1"),
+ Seq(s"beta${SECONDARY_INDEX_RECORD_KEY_SEPARATOR}row2"),
+ Seq(s"gamma${SECONDARY_INDEX_RECORD_KEY_SEPARATOR}row3")
+ )
+ assertCommonMetadataRecords(tablePath, expectedRecordIndexCount = 3,
includePartitionStats = true)
+
+ // Rollback must restore data and all MDT indexes.
+ rollback(metaClient, writeOptions, upsertInstant)
+ spark.catalog.refreshTable(tableName)
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+
+ checkAnswer(s"select id, rider, part from $tableName order by id")(
+ Seq("row1", "alpha", "p1"),
+ Seq("row2", "beta", "p1"),
+ Seq("row3", "gamma", "p2")
+ )
+ assertMetadataPartitions(metaClient, includePartitionStats = true,
includeSecondaryIndex = true)
+ assertCommonMetadataRecords(tablePath, expectedRecordIndexCount = 3,
includePartitionStats = true)
+ checkAnswer(s"select key from hudi_metadata('$tablePath') where type=7")(
+ Seq(s"alpha${SECONDARY_INDEX_RECORD_KEY_SEPARATOR}row1"),
+ Seq(s"beta${SECONDARY_INDEX_RECORD_KEY_SEPARATOR}row2"),
+ Seq(s"gamma${SECONDARY_INDEX_RECORD_KEY_SEPARATOR}row3")
+ )
+ }
+
+ @Test
+ def testTableVersionSixMetadataWritesAndRollback(): Unit = {
+ // Validate the legacy writer with the indexes supported by table version
6.
+ val tableName = "metadata_writer_v6"
+ val tablePath = s"$basePath/$tableName"
+ val writeOptions = metadataWriteOptions(tableName, streamingWrites =
false) +
+ (HoodieWriteConfig.WRITE_TABLE_VERSION.key() ->
HoodieTableVersion.SIX.versionCode().toString)
+
+ // Version 6 uses the legacy writer and supports a smaller index set.
+ createTable(tableName, tablePath, tableVersion =
Some(HoodieTableVersion.SIX.versionCode()))
+ spark.sql(
+ s"""insert into $tableName values
+ | (1, 'row1', 'alpha', 'p1'),
+ | (2, 'row2', 'beta', 'p2')
+ |""".stripMargin)
+
+ var metaClient = createMetaClient(tablePath)
+ val metadataMetaClient = createMetaClient(metaClient.getMetaPath +
"/metadata")
+ assertEquals(HoodieTableVersion.SIX,
metaClient.getTableConfig.getTableVersion)
+ assertEquals(HoodieTableVersion.SIX,
metadataMetaClient.getTableConfig.getTableVersion)
+ assertMetadataPartitions(metaClient, includePartitionStats = false,
includeSecondaryIndex = false)
+ assertCommonMetadataRecords(tablePath, expectedRecordIndexCount = 2,
includePartitionStats = false)
+
+ spark.sql(s"update $tableName set rider = 'updated', ts = 3 where id =
'row1'")
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+ val upsertInstant = metaClient.getActiveTimeline.getCommitsTimeline
+ .filterCompletedInstants.lastInstant().get().requestedTime()
+ checkAnswer(s"select id, rider from $tableName order by id")(
+ Seq("row1", "updated"),
+ Seq("row2", "beta")
+ )
+ assertCommonMetadataRecords(tablePath, expectedRecordIndexCount = 2,
includePartitionStats = false)
+
+ rollback(metaClient, writeOptions, upsertInstant)
+ spark.catalog.refreshTable(tableName)
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+
+ checkAnswer(s"select id, rider from $tableName order by id")(
+ Seq("row1", "alpha"),
+ Seq("row2", "beta")
+ )
+ assertEquals(HoodieTableVersion.SIX,
metaClient.getTableConfig.getTableVersion)
+ assertMetadataPartitions(metaClient, includePartitionStats = false,
includeSecondaryIndex = false)
+ assertCommonMetadataRecords(tablePath, expectedRecordIndexCount = 2,
includePartitionStats = false)
+
+ // Verify record-index deletion and bootstrap on the legacy table.
+ spark.sql(s"drop index record_index on $tableName")
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+
assertFalse(metaClient.getTableConfig.getMetadataPartitions.contains(MetadataPartitionType.RECORD_INDEX.getPartitionPath))
+ assertEquals(0L, spark.sql(s"select key from hudi_metadata('$tablePath')
where type=5").count())
+
+ spark.sql(s"create index record_index on $tableName (id)")
+ metaClient = HoodieTableMetaClient.reload(metaClient)
+
assertTrue(metaClient.getTableConfig.getMetadataPartitions.contains(MetadataPartitionType.RECORD_INDEX.getPartitionPath))
+ assertEquals(2L, spark.sql(s"select key from hudi_metadata('$tablePath')
where type=5").count())
+ }
+
+ private def createTable(
+ tableName: String,
+ tablePath: String,
+ tableVersion: Option[Int]): Unit = {
+ val tableVersionOption = tableVersion
+ .map(version => s"hoodie.write.table.version = '$version',")
+ .getOrElse("")
+ val streamingWrites = tableVersion.isEmpty
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | ts bigint,
+ | id string,
+ | rider string,
+ | part string
+ |) using hudi
+ | options (
+ | primaryKey = 'id',
+ | orderingFields = 'ts',
+ | type = 'cow',
+ | $tableVersionOption
+ | hoodie.metadata.enable = 'true',
+ | hoodie.metadata.index.column.stats.enable = 'true',
+ | hoodie.metadata.index.partition.stats.enable = 'true',
+ | hoodie.metadata.record.index.enable = 'true',
+ | hoodie.metadata.streaming.write.enabled = '$streamingWrites',
+ | hoodie.metadata.record.preparation.parallelism = '1',
+ | hoodie.metrics.on = 'true',
+ | hoodie.metrics.reporter.type = 'CONSOLE',
+ | hoodie.metrics.executor.enable = 'true',
+ | hoodie.datasource.write.recordkey.field = 'id',
+ | hoodie.datasource.write.partitionpath.field = 'part',
+ | hoodie.datasource.write.payload.class =
+ | 'org.apache.hudi.common.model.OverwriteWithLatestAvroPayload'
+ | )
+ | partitioned by (part)
+ | location '$tablePath'
+ |""".stripMargin)
+ }
+
+ private def metadataWriteOptions(tableName: String, streamingWrites:
Boolean): Map[String, String] = Map(
+ HoodieWriteConfig.TBL_NAME.key() -> tableName,
+ TABLE_TYPE.key() -> HoodieTableType.COPY_ON_WRITE.name(),
+ RECORDKEY_FIELD.key() -> "id",
+ PARTITIONPATH_FIELD.key() -> "part",
+ PRECOMBINE_FIELD.key() -> "ts",
+ HoodieMetadataConfig.ENABLE.key() -> "true",
+ HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key() -> "true",
+ HoodieMetadataConfig.ENABLE_METADATA_INDEX_PARTITION_STATS.key() -> "true",
+ HoodieMetadataConfig.GLOBAL_RECORD_LEVEL_INDEX_ENABLE_PROP.key() -> "true",
+ HoodieMetadataConfig.STREAMING_WRITE_ENABLED.key() ->
streamingWrites.toString,
+ HoodieMetadataConfig.RECORD_PREPARATION_PARALLELISM.key() -> "1",
+ "hoodie.metrics.on" -> "true",
+ "hoodie.metrics.reporter.type" -> "CONSOLE",
+ "hoodie.metrics.executor.enable" -> "true",
+ HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key() ->
classOf[InProcessLockProvider].getName
+ )
+
+ private def createMetaClient(tablePath: String): HoodieTableMetaClient =
+ HoodieTableMetaClient.builder()
+ .setBasePath(tablePath)
+ .setConf(HoodieTestUtils.getDefaultStorageConf)
+ .build()
+
+ private def rollback(
+ metaClient: HoodieTableMetaClient,
+ writeOptions: Map[String, String],
+ instantTime: String): Unit = {
+ val props = TypedProperties.fromMap(writeOptions.asJava)
+ val writeConfig = HoodieWriteConfig.newBuilder()
+ .withPath(metaClient.getBasePath)
+ .withSchema(new
TableSchemaResolver(metaClient).getTableSchema(false).toString)
+ .withProps(props)
+ .withEmbeddedTimelineServerEnabled(false)
+ .build()
+ val writeClient = new SparkRDDWriteClient(new
HoodieSparkEngineContext(jsc), writeConfig)
+ try {
+ assertTrue(writeClient.rollback(instantTime))
+ } finally {
+ writeClient.close()
+ }
+ }
+
+ private def assertMetadataPartitions(
+ metaClient: HoodieTableMetaClient,
+ includePartitionStats: Boolean,
+ includeSecondaryIndex: Boolean): Unit = {
+ val partitions = metaClient.getTableConfig.getMetadataPartitions
+
assertTrue(partitions.contains(MetadataPartitionType.FILES.getPartitionPath))
+
assertTrue(partitions.contains(MetadataPartitionType.COLUMN_STATS.getPartitionPath))
+
assertTrue(partitions.contains(MetadataPartitionType.RECORD_INDEX.getPartitionPath))
+ if (includePartitionStats) {
+
assertTrue(partitions.contains(MetadataPartitionType.PARTITION_STATS.getPartitionPath))
+ }
+ if (includeSecondaryIndex) {
+ assertTrue(partitions.contains("secondary_index_idx_rider"))
+ }
+ }
+
+ private def assertCommonMetadataRecords(
+ tablePath: String,
+ expectedRecordIndexCount: Long,
+ includePartitionStats: Boolean): Unit = {
+ // Payload types 3, 6, and 5 represent column stats, partition stats, and
record index.
+ assertTrue(spark.sql(s"select key from hudi_metadata('$tablePath') where
type=3").count() > 0)
+ if (includePartitionStats) {
+ assertTrue(spark.sql(s"select key from hudi_metadata('$tablePath') where
type=6").count() > 0)
+ }
+ assertEquals(
+ expectedRecordIndexCount,
+ spark.sql(s"select key from hudi_metadata('$tablePath') where
type=5").count())
+ }
+
+ private def checkAnswer(query: String)(expected: Seq[Any]*): Unit = {
+ val expectedRows = expected.map(_.mkString("|")).sorted.toList
+ val actualRows =
spark.sql(query).collect().map(_.toSeq.mkString("|")).sorted.toList
+ assertEquals(expectedRows, actualRows)
+ }
+}
