hudi-agent commented on code in PR #19731:
URL: https://github.com/apache/hudi/pull/19731#discussion_r4068875676
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriterTableVersionSix.java:
##########
@@ -299,36 +301,77 @@ void compactIfNecessary(BaseHoodieWriteClient<?,I,?,O>
writeClient, Option<Strin
// let's say we trigger compaction after C5 in MDT and so compaction
completes with C4001. but C5 crashed before completing in MDT.
// and again w/ C6, we will re-attempt compaction at which point latest
delta commit is C4 in MDT.
// and so we try compaction w/ instant C4001. So, we can avoid compaction
if we already have compaction w/ same instant time.
+ boolean compactionSchedulingHandled = false;
if
(metadataMetaClient.getActiveTimeline().filterCompletedInstants().containsInstant(compactionInstantTime))
{
LOG.info("Compaction with same {} time is already present in the
timeline.", compactionInstantTime);
- } else if (writeClient.scheduleCompactionAtInstant(compactionInstantTime,
Option.empty())) {
- LOG.info("Compaction is scheduled for timestamp {}",
compactionInstantTime);
- if (shouldDelegateToTableServiceManager(metadataWriteConfig,
ActionType.compaction)) {
- LOG.info("Skipping execution of compaction on MDT as it is delegated
to table service manager.");
- } else {
- writeClient.compact(compactionInstantTime, true);
+ return;
+ }
+
+ try {
+ if (request.includes(TableServiceType.COMPACT) &&
request.getMode().includesSchedule()) {
+ if (shouldDelegateScheduling(request, TableServiceType.COMPACT)) {
+ LOG.info("Skipping scheduling of compaction on MDT as it is
delegated to table service manager.");
+ compactionSchedulingHandled = true;
+ } else if
(writeClient.scheduleCompactionAtInstant(compactionInstantTime,
Option.empty())) {
+ LOG.info("Compaction is scheduled for timestamp {}",
compactionInstantTime);
+ compactionSchedulingHandled = true;
+ if (request.getMode().includesExecute()) {
+ if (shouldDelegateExecution(request, TableServiceType.COMPACT)) {
+ LOG.info("Skipping execution of compaction on MDT as it is
delegated to table service manager.");
+ } else {
+ writeClient.compact(compactionInstantTime, true);
+ }
+ }
+ }
}
- } else if (metadataWriteConfig.isLogCompactionEnabled()) {
- // Schedule and execute log compaction with suffixes based on the same
instant time. This ensures that any future
- // delta commits synced over will not have an instant time lesser than
the last completed instant on the
- // metadata table.
- final String logCompactionInstantTime =
createLogCompactionTimestamp(latestDeltaCommitTimeOpt.get());
- if
(metadataMetaClient.getActiveTimeline().filterCompletedInstants().containsInstant(logCompactionInstantTime))
{
- LOG.info("Log compaction with same {} time is already present in the
timeline.", logCompactionInstantTime);
- } else if
(writeClient.scheduleLogCompactionAtInstant(logCompactionInstantTime,
Option.empty())) {
- LOG.info("Log compaction is scheduled for timestamp {}",
logCompactionInstantTime);
- if (shouldDelegateToTableServiceManager(metadataWriteConfig,
ActionType.logcompaction)) {
- LOG.info("Skipping execution of log compaction on MDT as it is
delegated to table service manager.");
+ } catch (Exception e) {
+ metrics.ifPresent(m ->
m.incrementMetric(HoodieMetadataMetrics.COMPACTION_FAILURES, 1));
+ LOG.error("Error running compaction service in metadata table", e);
+ throw e;
+ }
+
+ // Preserve compaction priority when scheduling is delegated, but allow
log-compaction-only requests.
+ if (!compactionSchedulingHandled
+ && request.includes(TableServiceType.LOG_COMPACT)
+ && request.getMode().includesSchedule()
+ && metadataWriteConfig.isLogCompactionEnabled()) {
+ try {
+ if (shouldDelegateScheduling(request, TableServiceType.LOG_COMPACT)) {
+ LOG.info("Skipping scheduling of log compaction on MDT as it is
delegated to table service manager.");
} else {
- writeClient.logCompact(logCompactionInstantTime, true);
+ // Schedule and execute log compaction with suffixes based on the
same instant time. This ensures that any future
+ // delta commits synced over will not have an instant time lesser
than the last completed instant on the
+ // metadata table.
+ final String logCompactionInstantTime =
createLogCompactionTimestamp(latestDeltaCommitTimeOpt.get());
+ if
(metadataMetaClient.getActiveTimeline().filterCompletedInstants().containsInstant(logCompactionInstantTime))
{
+ LOG.info("Log compaction with same {} time is already present in
the timeline.", logCompactionInstantTime);
+ } else if
(writeClient.scheduleLogCompactionAtInstant(logCompactionInstantTime,
Option.empty())) {
+ LOG.info("Log compaction is scheduled for timestamp {}",
logCompactionInstantTime);
+ if (request.getMode().includesExecute()) {
+ if (shouldDelegateExecution(request,
TableServiceType.LOG_COMPACT)) {
+ LOG.info("Skipping execution of log compaction on MDT as it is
delegated to table service manager.");
+ } else {
+ writeClient.logCompact(logCompactionInstantTime, true);
+ }
+ }
+ }
}
+ } catch (Exception e) {
+ metrics.ifPresent(m ->
m.incrementMetric(HoodieMetadataMetrics.LOG_COMPACTION_FAILURES, 1));
+ LOG.error("Error running log compaction service in metadata table", e);
+ throw e;
}
}
}
@Override
- protected void executeClean(BaseHoodieWriteClient writeClient, String
instantTime) {
- writeClient.clean(createCleanTimestamp(instantTime));
+ protected boolean executeClean(BaseHoodieWriteClient writeClient,
Option<String> latestDeltaCommitTime) {
+ if (!latestDeltaCommitTime.isPresent()) {
+ LOG.warn("Skipping requested MDT clean for table version six: no
completed delta commit is available to derive the clean instant.");
+ return false;
+ }
+ writeClient.clean(createCleanTimestamp(latestDeltaCommitTime.get()));
Review Comment:
🤖 For v6 the clean instant is `latestDeltaCommit + 002`, and unlike the
compaction path there's no `containsInstant` check before scheduling. When the
tool is run repeatedly with no new data-table commits in between (idle table on
a cron), the second run's clean can produce a non-empty plan (run 1's
compaction created cleanable slices after run 1's clean already ran) and will
try to request a clean at an instant that already completed. Could you skip
when a completed clean with that timestamp is already on the MDT timeline,
mirroring the compaction check above?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableServicesTool.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * 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.utilities;
+
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.client.transaction.TransactionManager;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteConcurrencyMode;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieLockConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.metadata.HoodieTableMetadataWriter;
+import org.apache.hudi.metadata.MetadataTableServiceMode;
+import org.apache.hudi.metadata.MetadataTableServiceRequest;
+import org.apache.hudi.metadata.SparkMetadataWriterFactory;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.api.java.JavaSparkContext;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Standalone Spark tool for scheduling and executing metadata table services
directly through MDT writer APIs.
+ * This tool does not require an HTTP table-service-manager server. Users must
submit and schedule the job
+ * separately; enabling ingestion-side metadata table service delegation does
not launch it automatically.
+ */
+@Slf4j
+public class HoodieMetadataTableServicesTool {
+
+ private final Config cfg;
+ private final TypedProperties props;
+ private final HoodieSparkEngineContext engineContext;
+ private final HadoopStorageConfiguration storageConf;
+ private final HoodieTableMetaClient dataMetaClient;
+ private final Set<TableServiceType> services;
+ private final MetadataTableServiceMode mode;
+
+ public HoodieMetadataTableServicesTool(Config cfg, JavaSparkContext jsc) {
+ this(cfg, jsc, UtilHelpers.buildProperties(jsc.hadoopConfiguration(),
cfg.propsFilePath, cfg.configs));
+ }
+
+ HoodieMetadataTableServicesTool(Config cfg, JavaSparkContext jsc,
TypedProperties props) {
+ this.cfg = cfg;
+ this.props = props;
+ this.engineContext = new HoodieSparkEngineContext(jsc);
+ this.storageConf = new
HadoopStorageConfiguration(jsc.hadoopConfiguration());
+ this.dataMetaClient = HoodieTableMetaClient.builder()
+ .setConf(storageConf)
+ .setBasePath(cfg.basePath)
+ .build();
+ this.services = parseServices(cfg.services);
+ this.mode = MetadataTableServiceMode.fromValue(cfg.mode);
+ validateRequest(mode, services, cfg.instantTime);
+ }
+
+ public void run() {
+ dataMetaClient.reloadTableConfig();
+ if (!dataMetaClient.getTableConfig().isMetadataTableAvailable()) {
+ log.warn("Metadata table is not initialized for data table {}, skipping
table services", cfg.basePath);
+ return;
+ }
+
+ Set<TableServiceType> compactionServices = services.stream()
+ .filter(service -> service == TableServiceType.COMPACT || service ==
TableServiceType.LOG_COMPACT)
+ .collect(Collectors.toCollection(() ->
EnumSet.noneOf(TableServiceType.class)));
+
+
validateDataTableLockConfiguration(buildWriteConfig(WriteConcurrencyMode.SINGLE_WRITER));
+
+ // Finish plans left pending by a previous run before cleaning or
publishing new plans.
+ if (mode.includesExecute() && !compactionServices.isEmpty()) {
+ executeTableServicesPhase(compactionServices);
+ }
+
+ // Execute clean with the OCC writer so its own transaction manager
controls the required lock scope.
+ if (mode.includesExecute() && services.contains(TableServiceType.CLEAN)) {
+ executeTableServicesPhase(EnumSet.of(TableServiceType.CLEAN));
Review Comment:
🤖 Running clean through the OCC profile means the MDT client has the LAZY
failed-writes policy, so `writeClient.clean()` calls
`rollbackFailedWrites(mdtMetaClient)` *before* taking any lock and picks
inflight MDT deltacommits whose heartbeat is "expired". Ingestion writers use a
SINGLE_WRITER/EAGER MDT client (non-streaming) that never starts MDT
heartbeats, so a deltacommit that a data-table writer is committing right now
(under the DT lock) reads as expired (`getLastHeartbeatTime` → 0). The rollback
then waits for the lock, `resolveOrScheduleRollback` refetches from
`getCommitsTimeline()` (which includes completed instants) and
`validateRollbackCommitSequence` has no completed-instant guard — so could this
roll back a just-completed MDT deltacommit and leave the MDT missing that
commit's files? The same applies to an async `HoodieIndexer`'s long-running MDT
deltacommit (the EAGER branch excludes indexing commits, the LAZY branch does
not). @nsivabalan could you sanity-che
ck this interaction?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableServicesTool.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * 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.utilities;
+
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.client.transaction.TransactionManager;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteConcurrencyMode;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieLockConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.metadata.HoodieTableMetadataWriter;
+import org.apache.hudi.metadata.MetadataTableServiceMode;
+import org.apache.hudi.metadata.MetadataTableServiceRequest;
+import org.apache.hudi.metadata.SparkMetadataWriterFactory;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.api.java.JavaSparkContext;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Standalone Spark tool for scheduling and executing metadata table services
directly through MDT writer APIs.
+ * This tool does not require an HTTP table-service-manager server. Users must
submit and schedule the job
+ * separately; enabling ingestion-side metadata table service delegation does
not launch it automatically.
+ */
+@Slf4j
+public class HoodieMetadataTableServicesTool {
+
+ private final Config cfg;
+ private final TypedProperties props;
+ private final HoodieSparkEngineContext engineContext;
+ private final HadoopStorageConfiguration storageConf;
+ private final HoodieTableMetaClient dataMetaClient;
+ private final Set<TableServiceType> services;
+ private final MetadataTableServiceMode mode;
+
+ public HoodieMetadataTableServicesTool(Config cfg, JavaSparkContext jsc) {
+ this(cfg, jsc, UtilHelpers.buildProperties(jsc.hadoopConfiguration(),
cfg.propsFilePath, cfg.configs));
+ }
+
+ HoodieMetadataTableServicesTool(Config cfg, JavaSparkContext jsc,
TypedProperties props) {
+ this.cfg = cfg;
+ this.props = props;
+ this.engineContext = new HoodieSparkEngineContext(jsc);
+ this.storageConf = new
HadoopStorageConfiguration(jsc.hadoopConfiguration());
+ this.dataMetaClient = HoodieTableMetaClient.builder()
+ .setConf(storageConf)
+ .setBasePath(cfg.basePath)
+ .build();
+ this.services = parseServices(cfg.services);
+ this.mode = MetadataTableServiceMode.fromValue(cfg.mode);
+ validateRequest(mode, services, cfg.instantTime);
+ }
+
+ public void run() {
+ dataMetaClient.reloadTableConfig();
+ if (!dataMetaClient.getTableConfig().isMetadataTableAvailable()) {
+ log.warn("Metadata table is not initialized for data table {}, skipping
table services", cfg.basePath);
+ return;
+ }
+
+ Set<TableServiceType> compactionServices = services.stream()
+ .filter(service -> service == TableServiceType.COMPACT || service ==
TableServiceType.LOG_COMPACT)
+ .collect(Collectors.toCollection(() ->
EnumSet.noneOf(TableServiceType.class)));
+
+
validateDataTableLockConfiguration(buildWriteConfig(WriteConcurrencyMode.SINGLE_WRITER));
+
+ // Finish plans left pending by a previous run before cleaning or
publishing new plans.
+ if (mode.includesExecute() && !compactionServices.isEmpty()) {
+ executeTableServicesPhase(compactionServices);
+ }
+
+ // Execute clean with the OCC writer so its own transaction manager
controls the required lock scope.
+ if (mode.includesExecute() && services.contains(TableServiceType.CLEAN)) {
+ executeTableServicesPhase(EnumSet.of(TableServiceType.CLEAN));
+ }
+
+ if (mode.includesSchedule() && !compactionServices.isEmpty()) {
+ if (mode == MetadataTableServiceMode.SCHEDULE_AND_EXECUTE) {
+ // Preserve inline ordering: fully process compaction before log
compaction.
+ scheduleAndExecuteCompactionService(TableServiceType.COMPACT,
compactionServices);
+ scheduleAndExecuteCompactionService(TableServiceType.LOG_COMPACT,
compactionServices);
+ } else {
+ // Pure scheduling only publishes plans while holding the data-table
lock.
+ scheduleTableServicesPhase(compactionServices);
+ }
+ }
+
+ // Archive after all requested compaction services have completed.
+ if (mode.includesExecute() && services.contains(TableServiceType.ARCHIVE))
{
+ executeTableServicesPhase(EnumSet.of(TableServiceType.ARCHIVE));
+ }
+ }
+
+ private void scheduleAndExecuteCompactionService(TableServiceType service,
+ Set<TableServiceType>
requestedServices) {
+ if (!requestedServices.contains(service)) {
+ return;
+ }
+ Set<TableServiceType> serviceSet = EnumSet.of(service);
+ // Publish the plan under the data-table lock, then release it before
expensive OCC execution.
+ scheduleTableServicesPhase(serviceSet);
+ executeTableServicesPhase(serviceSet);
+ }
+
+ private void executeTableServicesPhase(Set<TableServiceType>
executionServices) {
+ // Let the OCC writer's transaction manager acquire the shared data-table
logical lock only where required.
+ // With no explicit instant, compaction services execute every pending
plan for the requested service types.
+ HoodieWriteConfig executionConfig =
buildWriteConfig(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL);
+ try (HoodieTableMetadataWriter writer = createWriter(executionConfig)) {
+ writer.executeTableServices(newRequest(MetadataTableServiceMode.EXECUTE,
executionServices));
+ } catch (Exception e) {
+ throw new HoodieException("Failed to execute metadata table services " +
executionServices, e);
+ }
+ }
+
+ private void scheduleTableServicesPhase(Set<TableServiceType>
schedulingServices) {
+ // The outer transaction owns the shared data-table lock; the
SINGLE_WRITER MDT writer must not reacquire it.
+ HoodieWriteConfig schedulingConfig =
buildWriteConfig(WriteConcurrencyMode.SINGLE_WRITER);
+ try (TransactionManager transactionManager =
createTransactionManager(schedulingConfig)) {
+ transactionManager.beginStateChange(Option.empty(), Option.empty());
+ // Close the writer before releasing the lock, preserving any primary
failure if either close fails.
+ try (AutoCloseable lock = () ->
transactionManager.endStateChange(Option.empty());
+ HoodieTableMetadataWriter writer = createWriter(schedulingConfig)) {
+
writer.scheduleTableServices(newRequest(MetadataTableServiceMode.SCHEDULE,
schedulingServices));
+ } catch (Exception e) {
+ throw new HoodieException("Failed to run lock-protected metadata table
services", e);
+ }
+ }
+ }
+
+ TransactionManager createTransactionManager(HoodieWriteConfig writeConfig) {
+ return new TransactionManager(writeConfig, dataMetaClient.getStorage());
+ }
+
+ HoodieTableMetadataWriter createWriter(HoodieWriteConfig writeConfig) {
+ HoodieTableMetadataWriter writer = SparkMetadataWriterFactory.create(
Review Comment:
🤖 Creating a fresh writer here runs the constructor's `initializeIfNeeded`,
and for v8+ `shouldInitializeFromFilesystem` returns true, so any index
partition enabled in the tool's config but absent from
`hoodie.table.metadata.partitions` (Spark defaults column stats and partition
stats to true) would be bootstrapped as a side effect — e.g. a Flink-ingested
table without column stats. In the schedule phase that build runs while holding
the data-table lock; in the execute phase it runs unlocked alongside DT
writers. Have you considered restricting the tool's enabled indexes to what the
table config already has (or failing fast if there is a mismatch)? I notice the
integration test explicitly disables column stats, which hints at this.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -1580,38 +1614,96 @@ public void performTableServices(Option<String>
inFlightInstantTimestamp, boolea
}
}
- static HoodieActiveTimeline
runPendingTableServicesOperationsAndRefreshTimeline(HoodieTableMetaClient
metadataMetaClient,
-
BaseHoodieWriteClient<?, ?, ?, ?> writeClient,
-
boolean initialTimelineRequiresRefresh,
-
Option<HoodieMetadataMetrics> metricsOption) {
+ @Override
+ public void scheduleTableServices(MetadataTableServiceRequest request) {
+ MetadataTableServiceRequest scheduleRequest =
request.copy(MetadataTableServiceMode.SCHEDULE);
+ runTableServicesInternal(scheduleRequest, true, false);
+ }
+
+ @Override
+ public void executeTableServices(MetadataTableServiceRequest request) {
+ MetadataTableServiceRequest executeRequest =
request.copy(MetadataTableServiceMode.EXECUTE);
+ runTableServicesInternal(executeRequest, true, false);
+ }
+
+ private boolean executePendingCompactionServices(MetadataTableServiceRequest
request,
+ HoodieActiveTimeline
activeTimeline,
+ BaseHoodieWriteClient<?, I,
?, O> writeClient) {
+ boolean ranServices = false;
try {
- HoodieActiveTimeline activeTimeline = initialTimelineRequiresRefresh ?
metadataMetaClient.reloadActiveTimeline() :
metadataMetaClient.getActiveTimeline();
- // finish off any pending log compaction or compactions operations if
any from previous attempt.
- boolean ranServices = false;
- if (activeTimeline.filterPendingCompactionTimeline().countInstants() >
0) {
- if
(writeClient.shouldDelegateToTableServiceManager(writeClient.getConfig(),
ActionType.compaction)) {
+ if (request.includes(TableServiceType.COMPACT)
+ && activeTimeline.filterPendingCompactionTimeline().countInstants()
> 0) {
+ if (shouldDelegateExecution(request, TableServiceType.COMPACT)) {
LOG.info("Skipping pending compactions on MDT as they are delegated
to table service manager.");
+ } else if (request.getInstantTime().isPresent()) {
+ writeClient.compact(request.getInstantTime().get(), true);
+ ranServices = true;
} else {
writeClient.runAnyPendingCompactions();
ranServices = true;
}
}
- if (activeTimeline.filterPendingLogCompactionTimeline().countInstants()
> 0) {
- if
(writeClient.shouldDelegateToTableServiceManager(writeClient.getConfig(),
ActionType.logcompaction)) {
+ if (request.includes(TableServiceType.LOG_COMPACT)
+ &&
activeTimeline.filterPendingLogCompactionTimeline().countInstants() > 0) {
+ if (shouldDelegateExecution(request, TableServiceType.LOG_COMPACT)) {
LOG.info("Skipping pending log compactions on MDT as they are
delegated to table service manager.");
+ } else if (request.getInstantTime().isPresent()) {
Review Comment:
🤖 nit: `shouldDelegateExecution`/`shouldDelegateScheduling`/`shouldDelegate`
plus `actionName`/`actionsProvider` add a fair bit of indirection for what's
essentially "look up a config string and check if it contains this action's
name." Consider whether `shouldDelegate` could just take the
`TableServiceManagerConfig` directly instead of a `Function<HoodieWriteConfig,
String>` provider, which would let readers skip a level of lambda-chasing.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableServicesTool.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * 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.utilities;
+
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.client.transaction.TransactionManager;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.model.TableServiceType;
+import org.apache.hudi.common.model.WriteConcurrencyMode;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieLockConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.metadata.HoodieTableMetadataWriter;
+import org.apache.hudi.metadata.MetadataTableServiceMode;
+import org.apache.hudi.metadata.MetadataTableServiceRequest;
+import org.apache.hudi.metadata.SparkMetadataWriterFactory;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hadoop.fs.Path;
+import org.apache.spark.api.java.JavaSparkContext;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Standalone Spark tool for scheduling and executing metadata table services
directly through MDT writer APIs.
+ * This tool does not require an HTTP table-service-manager server. Users must
submit and schedule the job
+ * separately; enabling ingestion-side metadata table service delegation does
not launch it automatically.
+ */
+@Slf4j
+public class HoodieMetadataTableServicesTool {
+
+ private final Config cfg;
+ private final TypedProperties props;
+ private final HoodieSparkEngineContext engineContext;
+ private final HadoopStorageConfiguration storageConf;
+ private final HoodieTableMetaClient dataMetaClient;
+ private final Set<TableServiceType> services;
+ private final MetadataTableServiceMode mode;
+
+ public HoodieMetadataTableServicesTool(Config cfg, JavaSparkContext jsc) {
+ this(cfg, jsc, UtilHelpers.buildProperties(jsc.hadoopConfiguration(),
cfg.propsFilePath, cfg.configs));
+ }
+
+ HoodieMetadataTableServicesTool(Config cfg, JavaSparkContext jsc,
TypedProperties props) {
+ this.cfg = cfg;
+ this.props = props;
+ this.engineContext = new HoodieSparkEngineContext(jsc);
+ this.storageConf = new
HadoopStorageConfiguration(jsc.hadoopConfiguration());
+ this.dataMetaClient = HoodieTableMetaClient.builder()
+ .setConf(storageConf)
+ .setBasePath(cfg.basePath)
+ .build();
+ this.services = parseServices(cfg.services);
+ this.mode = MetadataTableServiceMode.fromValue(cfg.mode);
+ validateRequest(mode, services, cfg.instantTime);
+ }
+
+ public void run() {
+ dataMetaClient.reloadTableConfig();
+ if (!dataMetaClient.getTableConfig().isMetadataTableAvailable()) {
+ log.warn("Metadata table is not initialized for data table {}, skipping
table services", cfg.basePath);
+ return;
+ }
+
+ Set<TableServiceType> compactionServices = services.stream()
+ .filter(service -> service == TableServiceType.COMPACT || service ==
TableServiceType.LOG_COMPACT)
+ .collect(Collectors.toCollection(() ->
EnumSet.noneOf(TableServiceType.class)));
+
+
validateDataTableLockConfiguration(buildWriteConfig(WriteConcurrencyMode.SINGLE_WRITER));
+
+ // Finish plans left pending by a previous run before cleaning or
publishing new plans.
+ if (mode.includesExecute() && !compactionServices.isEmpty()) {
+ executeTableServicesPhase(compactionServices);
+ }
+
+ // Execute clean with the OCC writer so its own transaction manager
controls the required lock scope.
+ if (mode.includesExecute() && services.contains(TableServiceType.CLEAN)) {
+ executeTableServicesPhase(EnumSet.of(TableServiceType.CLEAN));
+ }
+
+ if (mode.includesSchedule() && !compactionServices.isEmpty()) {
+ if (mode == MetadataTableServiceMode.SCHEDULE_AND_EXECUTE) {
+ // Preserve inline ordering: fully process compaction before log
compaction.
+ scheduleAndExecuteCompactionService(TableServiceType.COMPACT,
compactionServices);
+ scheduleAndExecuteCompactionService(TableServiceType.LOG_COMPACT,
compactionServices);
+ } else {
+ // Pure scheduling only publishes plans while holding the data-table
lock.
+ scheduleTableServicesPhase(compactionServices);
+ }
+ }
+
+ // Archive after all requested compaction services have completed.
+ if (mode.includesExecute() && services.contains(TableServiceType.ARCHIVE))
{
+ executeTableServicesPhase(EnumSet.of(TableServiceType.ARCHIVE));
+ }
+ }
+
+ private void scheduleAndExecuteCompactionService(TableServiceType service,
+ Set<TableServiceType>
requestedServices) {
+ if (!requestedServices.contains(service)) {
+ return;
+ }
+ Set<TableServiceType> serviceSet = EnumSet.of(service);
+ // Publish the plan under the data-table lock, then release it before
expensive OCC execution.
+ scheduleTableServicesPhase(serviceSet);
+ executeTableServicesPhase(serviceSet);
+ }
+
+ private void executeTableServicesPhase(Set<TableServiceType>
executionServices) {
+ // Let the OCC writer's transaction manager acquire the shared data-table
logical lock only where required.
+ // With no explicit instant, compaction services execute every pending
plan for the requested service types.
+ HoodieWriteConfig executionConfig =
buildWriteConfig(WriteConcurrencyMode.OPTIMISTIC_CONCURRENCY_CONTROL);
+ try (HoodieTableMetadataWriter writer = createWriter(executionConfig)) {
+ writer.executeTableServices(newRequest(MetadataTableServiceMode.EXECUTE,
executionServices));
Review Comment:
🤖 The execute phase runs `runAnyPendingCompactions()` without the data-table
lock (lock only at completion), while an ingestion writer's inline
`performTableServices` executes the same pending plan under the DT lock
(`doInitTable` → `executeUsingTxnManager`) unless the user has set
`hoodie.metadata.table.service.manager.actions` on the ingestion side. Nothing
in the tool verifies that, and both completing the same compaction instant
would leave duplicate base files / completed files for one instant. Is there a
cheap check the tool could do (e.g. read the last commit's extra metadata or
refuse to run without an explicit `--force`), or at least a loud warning in the
CLI help pointing to the README contract?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java:
##########
@@ -1531,30 +1534,61 @@ private SerializableFunction<HoodieRecord,
HoodieRecord> getRecordTaggerPartitio
*/
@Override
public void performTableServices(Option<String> inFlightInstantTimestamp,
boolean requiresTimelineRefresh) {
+ MetadataTableServiceRequest request =
MetadataTableServiceRequest.newBuilder()
+ .withMode(MetadataTableServiceMode.SCHEDULE_AND_EXECUTE)
+ .build();
+ runTableServicesInternal(request, requiresTimelineRefresh, true);
+ }
+
+ private void runTableServicesInternal(MetadataTableServiceRequest request,
+ boolean requiresTimelineRefresh,
boolean isInline) {
HoodieTimer metadataTableServicesTimer = HoodieTimer.start();
boolean allTableServicesExecutedSuccessfullyOrSkipped = true;
BaseHoodieWriteClient<?, I, ?, O> writeClient = getWriteClient();
try {
+ HoodieActiveTimeline activeTimeline = requiresTimelineRefresh
+ ? metadataMetaClient.reloadActiveTimeline() :
metadataMetaClient.getActiveTimeline();
// Run any pending table services operations and return the active
timeline
- HoodieActiveTimeline activeTimeline =
runPendingTableServicesOperationsAndRefreshTimeline(
- metadataMetaClient, writeClient, requiresTimelineRefresh, metrics);
+ if (request.getMode().includesExecute()
+ && executePendingCompactionServices(request, activeTimeline,
writeClient)) {
+ activeTimeline = metadataMetaClient.reloadActiveTimeline();
+ }
- Option<HoodieInstant> lastInstant =
activeTimeline.getDeltaCommitTimeline()
+ Option<String> latestDeltaCommitTime =
activeTimeline.getDeltaCommitTimeline()
.filterCompletedInstants()
- .lastInstant();
- if (!lastInstant.isPresent()) {
+ .lastInstant().map(HoodieInstant::requestedTime);
+ // Preserve inline behavior; explicit service requests need not depend
on a delta commit.
+ if (isInline && !latestDeltaCommitTime.isPresent()) {
+ LOG.warn("Skipping inline MDT maintenance after pending services: no
completed delta commit is available.");
return;
}
+
// Check and run clean operations.
- cleanIfNecessary(writeClient, lastInstant.get().requestedTime());
- // Do timeline validation before scheduling compaction/logCompaction
operations.
- if (validateCompactionScheduling(inFlightInstantTimestamp,
lastInstant.get().requestedTime())) {
- String latestDeltacommitTime = lastInstant.get().requestedTime();
- LOG.info("Latest deltacommit time found is {}, running compaction
operations.", latestDeltacommitTime);
- compactIfNecessary(writeClient, Option.of(latestDeltacommitTime));
+ if (request.getMode().includesExecute()
+ && request.includes(TableServiceType.CLEAN)
+ && !shouldDelegateExecution(request, TableServiceType.CLEAN)) {
+ runCleanService(writeClient, latestDeltaCommitTime);
+ }
+
+ if (request.getMode().includesSchedule()
+ && (request.includes(TableServiceType.COMPACT) ||
request.includes(TableServiceType.LOG_COMPACT))) {
+ if (!latestDeltaCommitTime.isPresent()) {
+ LOG.warn("Skipping requested MDT compaction/log-compaction
scheduling: no completed delta commit is available.");
+ } else if (validateCompactionScheduling(latestDeltaCommitTime.get())) {
+ // Do timeline validation before scheduling compaction/logCompaction
operations.
+ LOG.info("Latest delta commit time found is {}, scheduling
compaction operations.", latestDeltaCommitTime.get());
+ runCompactionServicesIfNecessary(writeClient, latestDeltaCommitTime,
request);
+ }
+ }
+
+ if (request.getMode().includesExecute() &&
request.includes(TableServiceType.ARCHIVE)) {
Review Comment:
🤖 nit: `isInline` is a bit ambiguous here since it also affects whether we
skip when no delta commit is present, not just "was this called from
performTableServices". Might be worth renaming to something like
`requiresCompletedDeltaCommit` to make the guard's intent clearer at the call
site.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
