This is an automated email from the ASF dual-hosted git repository.
codope 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 92771dbf5afe feat(client): add an opt-in to skip inline metadata table
services (#20350)
92771dbf5afe is described below
commit 92771dbf5afea049a0797c100a0d07e8e01815e5
Author: Sagar Sumit <[email protected]>
AuthorDate: Fri Oct 16 15:01:58 2026 +0530
feat(client): add an opt-in to skip inline metadata table services (#20350)
Adds hoodie.metadata.inline.table.services.enabled (default true). When
set to false, a data table writer does not run clean, compaction or
archival on the metadata table inline:
- Spark and Java: skipped when a write starts, under the table lock.
- Flink: skipped when the coordinator initializes the metadata table and
on each metadata table commit, the same places that already skip them
when streaming writes to the metadata table are enabled.
hoodie.table.services.enabled=false does not stop these, and is not
reused for this: writers that keep it off rely on these paths to compact
the metadata table, and data table archival cannot move past the latest
metadata table compaction.
Signed-off-by: codope <[email protected]>
---
.../hudi/client/HoodieFlinkTableServiceClient.java | 3 +-
.../FlinkHoodieBackedTableMetadataWriter.java | 3 +-
.../apache/hudi/client/HoodieJavaWriteClient.java | 4 +-
.../apache/hudi/client/SparkRDDWriteClient.java | 6 +-
.../TestSparkDeletePartitionTableServices.java | 118 +++++++++++++++++++++
.../hudi/common/config/HoodieMetadataConfig.java | 18 ++++
.../sink/TestStreamWriteOperatorCoordinator.java | 45 ++++++++
7 files changed, 193 insertions(+), 4 deletions(-)
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
index 579861677a14..8ec43d9781d4 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
@@ -206,7 +206,8 @@ public class HoodieFlinkTableServiceClient<T> extends
BaseHoodieTableServiceClie
}
// Do not perform table service for mdt when streaming write is
enabled, since the compaction/clean
// will be performed asynchronously in the dedicated compaction
pipeline.
- if (!this.config.getMetadataConfig().isStreamingWriteEnabled()) {
+ if (!this.config.getMetadataConfig().isStreamingWriteEnabled()
+ &&
this.config.getMetadataConfig().isInlineTableServicesEnabled()) {
metadataWriter.performTableServices(Option.empty());
}
}
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
index d1f55ae25f92..7deac80e1d0f 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/metadata/FlinkHoodieBackedTableMetadataWriter.java
@@ -142,7 +142,8 @@ public class FlinkHoodieBackedTableMetadataWriter extends
HoodieBackedTableMetad
// this method will be invoked during compaction of data table, here table
services for mdt is
// not performed if streaming writes mdt is enabled, since the
compaction/clean will be performed
// asynchronously in the dedicated compaction pipeline.
- if (!dataWriteConfig.getMetadataConfig().isStreamingWriteEnabled()) {
+ if (!dataWriteConfig.getMetadataConfig().isStreamingWriteEnabled()
+ && dataWriteConfig.getMetadataConfig().isInlineTableServicesEnabled())
{
performTableServices(Option.ofNullable(instantTime), false);
}
metadataMetaClient.reloadActiveTimeline();
diff --git
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java
index 18188ef23b5a..ebe36d5c748d 100644
---
a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java
+++
b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java
@@ -248,7 +248,9 @@ public class HoodieJavaWriteClient<T> extends
if (writer.hasPartitionsStateChanged()) {
metaClient.reloadTableConfig();
}
- writer.performTableServices(inFlightInstantTimestamp);
+ if (config.getMetadataConfig().isInlineTableServicesEnabled()) {
+ writer.performTableServices(inFlightInstantTimestamp);
+ }
}
} catch (Exception e) {
throw new HoodieException("Failed to instantiate Metadata table ", e);
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
index 35c9abc76719..c72cfc756082 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
@@ -395,7 +395,11 @@ public class SparkRDDWriteClient<T> extends
if (writer.hasPartitionsStateChanged()) {
metaClient.reloadTableConfig();
}
- writer.performTableServices(inFlightInstantTimestamp);
+ if (config.getMetadataConfig().isInlineTableServicesEnabled()) {
+ writer.performTableServices(inFlightInstantTimestamp);
+ } else {
+ log.info("Skipping metadata table services since inline metadata
table services are disabled");
+ }
}
} catch (Exception e) {
throw new HoodieException("Failed to instantiate Metadata table ", e);
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestSparkDeletePartitionTableServices.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestSparkDeletePartitionTableServices.java
new file mode 100644
index 000000000000..257fc9b7e320
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestSparkDeletePartitionTableServices.java
@@ -0,0 +1,118 @@
+/*
+ * 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.commit;
+
+import org.apache.hudi.client.HoodieWriteResult;
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteClientTestUtils;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.testutils.HoodieClientTestBase;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static
org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
+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.assertTrue;
+
+/**
+ * A delete runs metadata table services while bootstrapping the table, under
the table lock. Only the
+ * dedicated metadata knob turns that off; the general table services flag
alone must not, since writers
+ * that set it still rely on this path to compact the metadata table.
+ */
+public class TestSparkDeletePartitionTableServices extends
HoodieClientTestBase {
+
+ private static final String PARTITION =
HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
+
+ @Test
+ public void
leavesTheMetadataTableAloneWhenInlineMetadataServicesAreDisabled() {
+ List<HoodieInstant> before = metadataInstantsAfterAnInsert();
+
+ String deleteInstant = deletePartition(false);
+
+ // The delete's own deltacommit is the only change: no clean, compaction
or archival.
+ List<HoodieInstant> added = metadataInstants().stream()
+ .filter(i -> !before.contains(i)).collect(Collectors.toList());
+ assertEquals(1, added.size(), "unexpected metadata instants " + added);
+ assertEquals(DELTA_COMMIT_ACTION, added.get(0).getAction());
+ assertEquals(deleteInstant, added.get(0).requestedTime());
+ assertTrue(metadataInstants().containsAll(before), "nothing should have
been archived");
+ }
+
+ @Test
+ public void stillCompactsTheMetadataTableWhenOnlyTableServicesAreDisabled() {
+ List<HoodieInstant> before = metadataInstantsAfterAnInsert();
+
+ deletePartition(true);
+
+ // Also proves the setup makes compaction due, so the test above cannot
pass vacuously.
+ assertTrue(metadataInstants().stream()
+ .anyMatch(i -> i.getAction().equals(COMMIT_ACTION) &&
!before.contains(i)),
+ "a compaction was due and should have run");
+ }
+
+ private List<HoodieInstant> metadataInstantsAfterAnInsert() {
+ try (SparkRDDWriteClient client = getHoodieWriteClient(config(true,
true))) {
+ String instantTime = WriteClientTestUtils.createNewInstantTime();
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ List<HoodieRecord> records =
dataGen.generateInsertsForPartition(instantTime, 10, PARTITION);
+ client.commit(instantTime, client.insert(jsc.parallelize(records, 1),
instantTime));
+ }
+ return metadataInstants();
+ }
+
+ private String deletePartition(boolean inlineMetadataServices) {
+ try (SparkRDDWriteClient client = getHoodieWriteClient(config(false,
inlineMetadataServices))) {
+ String instantTime = WriteClientTestUtils.createNewInstantTime();
+ WriteClientTestUtils.startCommitWithTime(client, instantTime,
REPLACE_COMMIT_ACTION);
+ HoodieWriteResult result =
client.deletePartitions(Collections.singletonList(PARTITION), instantTime);
+ client.commit(instantTime, result.getWriteStatuses(), Option.empty(),
REPLACE_COMMIT_ACTION,
+ result.getPartitionToReplaceFileIds());
+ return instantTime;
+ }
+ }
+
+ private HoodieWriteConfig config(boolean tableServices, boolean
inlineMetadataServices) {
+ return getConfigBuilder()
+ .withTableServicesEnabled(tableServices)
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder()
+ .enable(true)
+ .withMaxNumDeltaCommitsBeforeCompaction(1)
+ .withInlineTableServicesEnabled(inlineMetadataServices)
+ .build())
+ .build();
+ }
+
+ private List<HoodieInstant> metadataInstants() {
+ return HoodieTestUtils.createMetaClient(storageConf,
HoodieTableMetadata.getMetadataTableBasePath(basePath))
+ .getActiveTimeline().getInstants();
+ }
+}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
index 6121bd250378..619fb7c55876 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java
@@ -100,6 +100,15 @@ public final class HoodieMetadataConfig extends
HoodieConfig {
+ "instead of executing them inline. This prevents the current
writer from executing compaction/logcompaction "
+ "on the metadata table, allowing a separate async pipeline to
handle them.");
+ public static final ConfigProperty<Boolean> INLINE_TABLE_SERVICES_ENABLED =
ConfigProperty
+ .key(METADATA_PREFIX + ".inline.table.services.enabled")
+ .defaultValue(true)
+ .markAdvanced()
+ .sinceVersion("1.3.0")
+ .withDocumentation("If false, a data table writer does not run clean,
compaction or archival on the metadata table "
+ + "inline. For jobs that must not maintain the metadata table, such
as a partition delete, "
+ + "and that leave it to the regular writers or a table service
platform.");
+
public static final Set<ActionType> SUPPORTED_TABLE_SERVICE_MANAGER_ACTIONS =
EnumSet.of(ActionType.compaction, ActionType.logcompaction);
@@ -746,6 +755,10 @@ public final class HoodieMetadataConfig extends
HoodieConfig {
return getBoolean(TABLE_SERVICE_MANAGER_ENABLED);
}
+ public boolean isInlineTableServicesEnabled() {
+ return getBoolean(INLINE_TABLE_SERVICES_ENABLED);
+ }
+
public String getTableServiceManagerActions() {
return getString(TABLE_SERVICE_MANAGER_ACTIONS);
}
@@ -1100,6 +1113,11 @@ public final class HoodieMetadataConfig extends
HoodieConfig {
return this;
}
+ public Builder withInlineTableServicesEnabled(boolean enabled) {
+ metadataConfig.setValue(INLINE_TABLE_SERVICES_ENABLED,
String.valueOf(enabled));
+ return this;
+ }
+
public Builder withTableServiceManagerActions(String actions) {
if (!actions.isEmpty()) {
validateTableServiceManagerActions(actions);
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
index 493282c2b139..1e68d9f2842e 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
@@ -22,6 +22,7 @@ import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.common.HoodieFlinkEngineContext;
import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
@@ -95,6 +96,7 @@ import org.slf4j.Logger;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.Field;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -104,6 +106,7 @@ import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
import static
org.apache.hudi.index.HoodieIndex.IndexType.GLOBAL_RECORD_LEVEL_INDEX;
@@ -834,6 +837,48 @@ public class TestStreamWriteOperatorCoordinator {
completedTimeline.nthFromLastInstant(3).get(), is(compactionInstant));
}
+ @Test
+ void testSkipInlineMetadataTableServices() throws Exception {
+ coordinator.close();
+ reset();
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ int metadataCompactionDeltaCommits = 5;
+ conf.set(FlinkOptions.METADATA_ENABLED, true);
+ conf.set(FlinkOptions.METADATA_COMPACTION_DELTA_COMMITS,
metadataCompactionDeltaCommits);
+ conf.setString(HoodieMetadataConfig.INLINE_TABLE_SERVICES_ENABLED.key(),
"false");
+ coordinator = startCoordinator(conf, 1);
+
+ final String metadataTableBasePath =
HoodieTableMetadata.getMetadataTableBasePath(tempFile.getAbsolutePath());
+ HoodieTableMetaClient metadataTableMetaClient =
HoodieTestUtils.createMetaClient(
+ new
HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(conf)),
metadataTableBasePath);
+ List<HoodieInstant> before =
metadataTableMetaClient.getActiveTimeline().getInstants();
+
+ // Commits past the compaction threshold only add their own deltacommits:
no clean, compaction or archival.
+ List<String> instants = new ArrayList<>();
+ for (int ckp = 0; ckp <= metadataCompactionDeltaCommits; ckp++) {
+ instants.add(mockWriteWithMetadata(ckp));
+ }
+ List<HoodieInstant> afterWrites =
metadataTableMetaClient.reloadActiveTimeline().getInstants();
+ assertTrue(afterWrites.containsAll(before), "nothing should have been
archived");
+ List<HoodieInstant> added = afterWrites.stream().filter(i ->
!before.contains(i)).collect(Collectors.toList());
+ assertTrue(added.stream().allMatch(i ->
i.getAction().equals(HoodieTimeline.DELTA_COMMIT_ACTION)),
+ "unexpected metadata instants " + added);
+ assertEquals(instants,
added.stream().map(HoodieInstant::requestedTime).collect(Collectors.toList()));
+
+ // Restarting the coordinator does not run them either.
+ coordinator.close();
+ coordinator = startCoordinator(conf, 1);
+ assertEquals(afterWrites,
metadataTableMetaClient.reloadActiveTimeline().getInstants());
+
+ // The same restart with the services enabled compacts, so the checks
above cannot pass vacuously.
+ coordinator.close();
+ conf.setString(HoodieMetadataConfig.INLINE_TABLE_SERVICES_ENABLED.key(),
"true");
+ coordinator = startCoordinator(conf, 1);
+
assertTrue(metadataTableMetaClient.reloadActiveTimeline().getInstants().stream()
+ .anyMatch(i -> i.getAction().equals(HoodieTimeline.COMMIT_ACTION)
&& !afterWrites.contains(i)),
+ "a compaction was due and should have run");
+ }
+
@Test
void testSyncMetadataTableWithLogCompaction() throws Exception {
// reset
