feat(metadata): Allow users to safely execute compaction plans on metadata table concurrently through a table service platform (rather than only inline during write)

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieLockConfig.java

@@ -20,12 +20,19 @@
 import org.apache.hudi.client.transaction.BucketIndexConcurrentFileWritesConflictResolutionStrategy;
 import org.apache.hudi.client.transaction.ConflictResolutionStrategy;
 import org.apache.hudi.client.transaction.SimpleConcurrentFileWritesConflictResolutionStrategy;
+import org.apache.hudi.client.transaction.lock.FileSystemBasedLockProvider;
+import org.apache.hudi.client.transaction.lock.StorageBasedLockProvider;
+import org.apache.hudi.client.transaction.lock.ZookeeperBasedImplicitBasePathLockProvider;
+import org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider;
 import org.apache.hudi.common.config.ConfigClassProperty;
 import org.apache.hudi.common.config.ConfigGroups;
 import org.apache.hudi.common.config.ConfigProperty;
 import org.apache.hudi.common.config.HoodieConfig;
+import org.apache.hudi.common.config.LockConfiguration;
+import org.apache.hudi.common.config.TypedProperties;
 import org.apache.hudi.common.lock.LockProvider;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.index.HoodieIndex;
 
 import java.io.File;
@@ -241,6 +248,17 @@ public class HoodieLockConfig extends HoodieConfig {
   @Deprecated
   public static final String LOCK_PROVIDER_CLASS_PROP = LOCK_PROVIDER_CLASS_NAME.key();
 
+  // Lock provider class names from modules not directly accessible in hudi-client-common.
+  public static final String HIVE_METASTORE_BASED_LOCK_PROVIDER_CLASS =
+      "org.apache.hudi.hive.transaction.lock.HiveMetastoreBasedLockProvider";
+  public static final String DYNAMODB_BASED_LOCK_PROVIDER_CLASS =
+      "org.apache.hudi.aws.transaction.lock.DynamoDBBasedLockProvider";
+  public static final String DYNAMODB_BASED_IMPLICIT_PARTITION_KEY_LOCK_PROVIDER_CLASS =
+      "org.apache.hudi.aws.transaction.lock.DynamoDBBasedImplicitPartitionKeyLockProvider";
+  private static final String DYNAMODB_BASED_LOCK_PROPERTY_PREFIX = LOCK_PREFIX + "dynamodb.";
+  private static final String DYNAMODB_LOCK_PARTITION_KEY_PROP_KEY = DYNAMODB_BASED_LOCK_PROPERTY_PREFIX + "partition_key";
+  private static final String STORAGE_BASED_LOCK_PROPERTY_PREFIX = LOCK_PREFIX + "storage.";
+
   private HoodieLockConfig() {
     super();
   }
@@ -249,6 +267,109 @@ public static HoodieLockConfig.Builder newBuilder() {
     return new HoodieLockConfig.Builder();
   }
 
+  /**
+   * Derive a {@link HoodieLockConfig} for a different table by copying lock-related configs from the
+   * given write config. Only built-in lock providers with explicitly set lock keys/paths are supported;
+   * providers that implicitly derive their lock identity from the table's base path are rejected.
+   *
+   * @param lockProviderClass the lock provider class name
+   * @param writeConfig       the source write config to copy lock properties from
+   * @return a new HoodieLockConfig with the relevant lock properties
+   * @throws IllegalArgumentException if required lock keys are missing or the provider derives its lock identity implicitly
+   * @throws HoodieException if the lock provider is not a built-in provider
+   */
+  public static HoodieLockConfig deriveLockConfigForDifferentTable(String lockProviderClass, HoodieWriteConfig writeConfig) {
+    TypedProperties dataProps = writeConfig.getProps();
+    Properties lockProps = new Properties();
+    lockProps.put(LOCK_PROVIDER_CLASS_NAME.key(), lockProviderClass);
+
+    // Common lock configs used by all providers
+    lockProps.put(LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS.key(),
+        writeConfig.getStringOrDefault(LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS));
+    lockProps.put(LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS.key(),
+        writeConfig.getStringOrDefault(LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS));
+    lockProps.put(LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS.key(),
+        writeConfig.getStringOrDefault(LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS));
+    lockProps.put(LOCK_ACQUIRE_NUM_RETRIES.key(),
+        writeConfig.getStringOrDefault(LOCK_ACQUIRE_NUM_RETRIES));
+    lockProps.put(LOCK_ACQUIRE_CLIENT_NUM_RETRIES.key(),
+        writeConfig.getStringOrDefault(LOCK_ACQUIRE_CLIENT_NUM_RETRIES));
+    lockProps.put(LOCK_ACQUIRE_WAIT_TIMEOUT_MS.key(),
+        String.valueOf(writeConfig.getIntOrDefault(LOCK_ACQUIRE_WAIT_TIMEOUT_MS)));
+    lockProps.put(LOCK_HEARTBEAT_INTERVAL_MS.key(),
+        String.valueOf(writeConfig.getIntOrDefault(LOCK_HEARTBEAT_INTERVAL_MS)));
+
+    // Provider-specific configs with lock key validation.
+    // Providers whose lock identity (key/path) is inferred from table name or base path at build time
+    // won't carry those inferred values into a rebuilt config for a different table, so we either
+    // require them to be set explicitly or reject providers that offer no explicit override.
+    if (FileSystemBasedLockProvider.class.getCanonicalName().equals(lockProviderClass)) {
+      copyPropsWithPrefix(dataProps, lockProps, LockConfiguration.FILESYSTEM_BASED_LOCK_PROPERTY_PREFIX);
+      if (!lockProps.containsKey(LockConfiguration.FILESYSTEM_LOCK_PATH_PROP_KEY)) {
+        throw new IllegalArgumentException(LockConfiguration.FILESYSTEM_LOCK_PATH_PROP_KEY
+            + " must be explicitly set when using " + lockProviderClass
+            + ". Without it, the lock path is derived from the table's base path,"
+            + " which would resolve to a different location when building a lock config for a different table.");
+      }
+    } else if (ZookeeperBasedLockProvider.class.getCanonicalName().equals(lockProviderClass)) {
+      copyPropsWithPrefix(dataProps, lockProps, LockConfiguration.ZOOKEEPER_BASED_LOCK_PROPERTY_PREFIX);
+      if (!lockProps.containsKey(LockConfiguration.ZK_LOCK_KEY_PROP_KEY)) {
+        throw new IllegalArgumentException(LockConfiguration.ZK_LOCK_KEY_PROP_KEY
+            + " must be explicitly set when using " + lockProviderClass
+            + ". The inferred default from table name is not propagated"
+            + " when building a lock config for a different table.");
+      }
+    } else if (ZookeeperBasedImplicitBasePathLockProvider.class.getCanonicalName().equals(lockProviderClass)) {
+      throw new IllegalArgumentException(lockProviderClass
+          + " is not supported because it derives its lock identity from the table's base path,"
+          + " which would resolve to a different lock when building a lock config for a different table."
+          + " Use " + ZookeeperBasedLockProvider.class.getCanonicalName() + " with an explicit "
+          + LockConfiguration.ZK_LOCK_KEY_PROP_KEY + " instead.");
+    } else if (HIVE_METASTORE_BASED_LOCK_PROVIDER_CLASS.equals(lockProviderClass)) {
+      copyPropsWithPrefix(dataProps, lockProps, LockConfiguration.HIVE_METASTORE_LOCK_PROPERTY_PREFIX);
+      if (!lockProps.containsKey(LockConfiguration.HIVE_DATABASE_NAME_PROP_KEY)) {
+        throw new IllegalArgumentException(LockConfiguration.HIVE_DATABASE_NAME_PROP_KEY
+            + " must be explicitly set when using " + lockProviderClass);
+      }
+      if (!lockProps.containsKey(LockConfiguration.HIVE_TABLE_NAME_PROP_KEY)) {
+        throw new IllegalArgumentException(LockConfiguration.HIVE_TABLE_NAME_PROP_KEY
+            + " must be explicitly set when using " + lockProviderClass);
+      }
+    } else if (DYNAMODB_BASED_LOCK_PROVIDER_CLASS.equals(lockProviderClass)) {
+      copyPropsWithPrefix(dataProps, lockProps, DYNAMODB_BASED_LOCK_PROPERTY_PREFIX);
+      if (!lockProps.containsKey(DYNAMODB_LOCK_PARTITION_KEY_PROP_KEY)) {
+        throw new IllegalArgumentException(DYNAMODB_LOCK_PARTITION_KEY_PROP_KEY
+            + " must be explicitly set when using " + lockProviderClass
+            + ". The inferred default from table name is not propagated"
+            + " when building a lock config for a different table.");
+      }
+    } else if (DYNAMODB_BASED_IMPLICIT_PARTITION_KEY_LOCK_PROVIDER_CLASS.equals(lockProviderClass)) {
+      throw new IllegalArgumentException(lockProviderClass
+          + " is not supported because it derives its lock identity from the table's base path,"
+          + " which would resolve to a different lock when building a lock config for a different table."
+          + " Use " + DYNAMODB_BASED_LOCK_PROVIDER_CLASS + " with an explicit "
+          + DYNAMODB_LOCK_PARTITION_KEY_PROP_KEY + " instead.");
+    } else if (StorageBasedLockProvider.class.getCanonicalName().equals(lockProviderClass)) {
+      throw new IllegalArgumentException(lockProviderClass
+          + " is not supported because it derives its lock identity from the table's base path,"
+          + " which would resolve to a different lock when building a lock config for a different table,"
+          + " and does not provide an explicit lock path override.");
+    } else {
+      throw new HoodieException(writeConfig.getWriteConcurrencyMode()
+          + " is only supported for built-in lock providers. Unsupported lock provider: " + lockProviderClass);
+    }
+
+    return newBuilder().fromProperties(lockProps).build();
+  }
+
+  private static void copyPropsWithPrefix(TypedProperties source, Properties target, String prefix) {
+    for (String key : source.stringPropertyNames()) {
+      if (key.startsWith(prefix)) {
+        target.setProperty(key, source.getProperty(key));
+      }
+    }
+  }
+
   public static class Builder {
 
     private final HoodieLockConfig lockConfig = new HoodieLockConfig();

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java

@@ -25,9 +25,11 @@
 import org.apache.hudi.avro.model.HoodieRestorePlan;
 import org.apache.hudi.avro.model.HoodieRollbackMetadata;
 import org.apache.hudi.client.BaseHoodieWriteClient;
+import org.apache.hudi.client.RunsTableService;
 import org.apache.hudi.client.WriteStatus;
 import org.apache.hudi.common.config.HoodieConfig;
 import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.ActionType;
 import org.apache.hudi.common.data.HoodieData;
 import org.apache.hudi.common.engine.EngineType;
 import org.apache.hudi.common.engine.HoodieEngineContext;
@@ -156,7 +158,7 @@
  *
  * @param <I> Type of input for the write client
  */
-public abstract class HoodieBackedTableMetadataWriter<I, O> implements HoodieTableMetadataWriter<I, O> {
+public abstract class HoodieBackedTableMetadataWriter<I, O> implements HoodieTableMetadataWriter<I, O>, RunsTableService {
 
   static final Logger LOG = LoggerFactory.getLogger(HoodieBackedTableMetadataWriter.class);
 
@@ -2153,12 +2155,20 @@ static HoodieActiveTimeline runPendingTableServicesOperationsAndRefreshTimeline(
       // finish off any pending log compaction or compactions operations if any from previous attempt.
       boolean ranServices = false;
       if (activeTimeline.filterPendingCompactionTimeline().countInstants() > 0) {
-        writeClient.runAnyPendingCompactions();
-        ranServices = true;
+        if (writeClient.shouldDelegateToTableServiceManager(writeClient.getConfig(), ActionType.compaction)) {
+          LOG.info("Skipping pending compactions on MDT as they are delegated to table service manager.");
+        } else {
+          writeClient.runAnyPendingCompactions();
+          ranServices = true;
+        }
       }
       if (activeTimeline.filterPendingLogCompactionTimeline().countInstants() > 0) {
-        writeClient.runAnyPendingLogCompactions();
-        ranServices = true;
+        if (writeClient.shouldDelegateToTableServiceManager(writeClient.getConfig(), ActionType.logcompaction)) {
+          LOG.info("Skipping pending log compactions on MDT as they are delegated to table service manager.");
+        } else {
+          writeClient.runAnyPendingLogCompactions();
+          ranServices = true;
+        }
       }
       return ranServices ? metadataMetaClient.reloadActiveTimeline() : activeTimeline;
     } catch (Exception e) {
@@ -2201,7 +2211,11 @@ void compactIfNecessary(BaseHoodieWriteClient<?,I,?,O> writeClient, Option<Strin
         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);
-        writeClient.compact(compactionInstantTime, true);
+        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);
+        }
       }
     } catch (Exception e) {
       metrics.ifPresent(m -> m.incrementMetric(HoodieMetadataMetrics.COMPACTION_FAILURES, 1));
@@ -2217,7 +2231,11 @@ void compactIfNecessary(BaseHoodieWriteClient<?,I,?,O> writeClient, Option<Strin
         Option<String> scheduledLogCompaction = writeClient.scheduleLogCompaction(Option.empty());
         if (scheduledLogCompaction.isPresent()) {
           LOG.info("Log compaction is scheduled for timestamp {}", scheduledLogCompaction.get());
-          writeClient.logCompact(scheduledLogCompaction.get(), true);
+          if (shouldDelegateToTableServiceManager(metadataWriteConfig, ActionType.logcompaction)) {
+            LOG.info("Skipping execution of log compaction on MDT as it is delegated to table service manager.");
+          } else {
+            writeClient.logCompact(scheduledLogCompaction.get(), true);
+          }
         }
       }
     } catch (Exception e) {

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriterTableVersionSix.java

@@ -21,6 +21,7 @@
 import org.apache.hudi.avro.model.HoodieRollbackMetadata;
 import org.apache.hudi.client.BaseHoodieWriteClient;
 import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.ActionType;
 import org.apache.hudi.common.engine.HoodieEngineContext;
 import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -289,7 +290,11 @@ void compactIfNecessary(BaseHoodieWriteClient<?,I,?,O> writeClient, Option<Strin
       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);
-      writeClient.compact(compactionInstantTime, true);
+      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);
+      }
     } 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
@@ -299,7 +304,11 @@ void compactIfNecessary(BaseHoodieWriteClient<?,I,?,O> writeClient, Option<Strin
         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);
-        writeClient.logCompact(logCompactionInstantTime, true);
+        if (shouldDelegateToTableServiceManager(metadataWriteConfig, ActionType.logcompaction)) {
+          LOG.info("Skipping execution of log compaction on MDT as it is delegated to table service manager.");
+        } else {
+          writeClient.logCompact(logCompactionInstantTime, true);
+        }
       }
     }
   }