You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "nsivabalan (via GitHub)" <gi...@apache.org> on 2026/03/16 20:56:28 UTC
Re: [PR] 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]
nsivabalan commented on code in PR #18295:
URL: https://github.com/apache/hudi/pull/18295#discussion_r2942915009
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java:
##########
@@ -362,6 +387,99 @@ public static HoodieWriteConfig createMetadataWriteConfig(
return metadataWriteConfig;
}
+ /**
+ * Build the lock config for the metadata table by extracting lock-specific properties from the
+ * data table's write config. This avoids copying all properties (which would overwrite MDT-specific
+ * settings like base path and auto-clean).
+ */
+ private static HoodieLockConfig buildMetadataLockConfig(HoodieWriteConfig writeConfig) {
+ TypedProperties props = writeConfig.getProps();
+ HoodieLockConfig.Builder lockConfigBuilder = HoodieLockConfig.newBuilder()
+ .withClientNumRetries(Integer.parseInt(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_NUM_RETRIES.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_NUM_RETRIES.defaultValue())))
+ .withClientRetryWaitTimeInMillis(Long.parseLong(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS.defaultValue())))
+ .withLockWaitTimeInMillis(Long.valueOf(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_WAIT_TIMEOUT_MS.key(),
+ String.valueOf(HoodieLockConfig.LOCK_ACQUIRE_WAIT_TIMEOUT_MS.defaultValue()))))
+ .withNumRetries(Integer.parseInt(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_NUM_RETRIES.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_NUM_RETRIES.defaultValue())))
+ .withRetryWaitTimeInMillis(Long.parseLong(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS.defaultValue())))
+ .withRetryMaxWaitTimeInMillis(Long.parseLong(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS.defaultValue())))
+ .withHeartbeatIntervalInMillis(Long.valueOf(props.getString(
+ HoodieLockConfig.LOCK_HEARTBEAT_INTERVAL_MS.key(),
+ String.valueOf(HoodieLockConfig.LOCK_HEARTBEAT_INTERVAL_MS.defaultValue()))));
+
+ String lockProviderClass = writeConfig.getLockProviderClass();
+ if (lockProviderClass == null) {
+ return lockConfigBuilder.build();
+ }
+
+ Properties providerProp = new Properties();
+ providerProp.setProperty(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key(), lockProviderClass);
+ lockConfigBuilder.fromProperties(providerProp);
+
+ if (ZookeeperBasedLockProvider.class.getName().equals(lockProviderClass)) {
+ if (props.containsKey(HoodieLockConfig.ZK_CONNECT_URL.key())) {
+ lockConfigBuilder.withZkQuorum(props.getString(HoodieLockConfig.ZK_CONNECT_URL.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.ZK_BASE_PATH.key())) {
+ lockConfigBuilder.withZkBasePath(props.getString(HoodieLockConfig.ZK_BASE_PATH.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.ZK_LOCK_KEY.key())) {
+ lockConfigBuilder.withZkLockKey(props.getString(HoodieLockConfig.ZK_LOCK_KEY.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.ZK_PORT.key())) {
+ lockConfigBuilder.withZkPort(props.getString(HoodieLockConfig.ZK_PORT.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.ZK_SESSION_TIMEOUT_MS.key())) {
+ lockConfigBuilder.withZkSessionTimeoutInMs(
+ Long.valueOf(props.getString(HoodieLockConfig.ZK_SESSION_TIMEOUT_MS.key())));
+ }
+ if (props.containsKey(HoodieLockConfig.ZK_CONNECTION_TIMEOUT_MS.key())) {
+ lockConfigBuilder.withZkConnectionTimeoutInMs(
+ Long.valueOf(props.getString(HoodieLockConfig.ZK_CONNECTION_TIMEOUT_MS.key())));
+ }
+ } else if (FileSystemBasedLockProvider.class.getName().equals(lockProviderClass)) {
+ if (props.containsKey(HoodieLockConfig.FILESYSTEM_LOCK_PATH.key())) {
+ lockConfigBuilder.withFileSystemLockPath(props.getString(HoodieLockConfig.FILESYSTEM_LOCK_PATH.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.FILESYSTEM_LOCK_EXPIRE.key())) {
+ lockConfigBuilder.withFileSystemLockExpire(
+ Integer.parseInt(props.getString(HoodieLockConfig.FILESYSTEM_LOCK_EXPIRE.key())));
+ }
+ } else if (lockProviderClass.contains("HiveMetastoreBasedLockProvider")) {
+ if (props.containsKey(HoodieLockConfig.HIVE_DATABASE_NAME.key())) {
+ lockConfigBuilder.withHiveDatabaseName(props.getString(HoodieLockConfig.HIVE_DATABASE_NAME.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.HIVE_TABLE_NAME.key())) {
+ lockConfigBuilder.withHiveTableName(props.getString(HoodieLockConfig.HIVE_TABLE_NAME.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.HIVE_METASTORE_URI.key())) {
+ lockConfigBuilder.withHiveMetastoreURIs(props.getString(HoodieLockConfig.HIVE_METASTORE_URI.key()));
+ }
+ } else {
+ // For any custom lock provider, pass through all lock-prefixed properties
+ // so provider-specific configs are preserved.
Review Comment:
bcoz, if a new lock provider is added tomorrow, how do we ensure this code is updated?
for eg, I already don't see dynamo db, storage based lock provider not being accounted here.
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java:
##########
@@ -81,6 +82,28 @@ public final class HoodieMetadataConfig extends HoodieConfig {
+ "in streaming manner rather than two disjoint writes. By default "
+ "streaming writes to metadata table is enabled for SPARK engine for incremental operations and disabled for all other cases.");
+ public static final ConfigProperty<String> METADATA_WRITE_CONCURRENCY_MODE = ConfigProperty
+ .key(METADATA_PREFIX + ".write.concurrency.mode")
Review Comment:
can we add tests to TestHoodieMetadataConfig
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java:
##########
@@ -81,6 +82,28 @@ public final class HoodieMetadataConfig extends HoodieConfig {
+ "in streaming manner rather than two disjoint writes. By default "
+ "streaming writes to metadata table is enabled for SPARK engine for incremental operations and disabled for all other cases.");
+ public static final ConfigProperty<String> METADATA_WRITE_CONCURRENCY_MODE = ConfigProperty
+ .key(METADATA_PREFIX + ".write.concurrency.mode")
+ .defaultValue(WriteConcurrencyMode.SINGLE_WRITER.name())
+ .markAdvanced()
+ .withDocumentation("Change this to OPTIMISTIC_CONCURRENCY_CONTROL when MDT operations are being performed "
+ + "from an external concurrent writer (such as a table service platform) so that appropriate locks are taken.");
+
+ public static final ConfigProperty<Boolean> TABLE_SERVICE_MANAGER_ENABLED = ConfigProperty
+ .key(METADATA_PREFIX + ".table.service.manager.enabled")
Review Comment:
I see why you have taken this route.
we are re-using TABLE_SERVICE_MANAGER_ENABLED and TABLE_SERVICE_MANAGER_ACTIONS from HoodieTableServiceManagerConfig.
In that case, we can retain the two configs.
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java:
##########
@@ -81,6 +82,28 @@ public final class HoodieMetadataConfig extends HoodieConfig {
+ "in streaming manner rather than two disjoint writes. By default "
+ "streaming writes to metadata table is enabled for SPARK engine for incremental operations and disabled for all other cases.");
+ public static final ConfigProperty<String> METADATA_WRITE_CONCURRENCY_MODE = ConfigProperty
+ .key(METADATA_PREFIX + ".write.concurrency.mode")
+ .defaultValue(WriteConcurrencyMode.SINGLE_WRITER.name())
+ .markAdvanced()
+ .withDocumentation("Change this to OPTIMISTIC_CONCURRENCY_CONTROL when MDT operations are being performed "
+ + "from an external concurrent writer (such as a table service platform) so that appropriate locks are taken.");
+
+ public static final ConfigProperty<Boolean> TABLE_SERVICE_MANAGER_ENABLED = ConfigProperty
+ .key(METADATA_PREFIX + ".table.service.manager.enabled")
Review Comment:
how about folding two configs into just 1.
`hoodie.metadata.skip.table.service.executions`
There won't be any default value.
value could be comma separated list of table service actions. like `compaction`, `logcompaction` and `clean`.
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriterTableVersionSix.java:
##########
@@ -279,7 +286,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 (tsmConfig.isEnabledAndActionSupported(ActionType.logcompaction)) {
+ LOG.info("Skipping execution of log compaction on MDT as it is delegated to table service manager.");
+ } else {
+ writeClient.logCompact(logCompactionInstantTime, true);
+ }
Review Comment:
looks like we are only supporting compaction and log compactions for now.
can we call out in the config property documentation for the new configs added
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java:
##########
@@ -362,6 +387,99 @@ public static HoodieWriteConfig createMetadataWriteConfig(
return metadataWriteConfig;
}
+ /**
+ * Build the lock config for the metadata table by extracting lock-specific properties from the
+ * data table's write config. This avoids copying all properties (which would overwrite MDT-specific
+ * settings like base path and auto-clean).
+ */
+ private static HoodieLockConfig buildMetadataLockConfig(HoodieWriteConfig writeConfig) {
+ TypedProperties props = writeConfig.getProps();
+ HoodieLockConfig.Builder lockConfigBuilder = HoodieLockConfig.newBuilder()
+ .withClientNumRetries(Integer.parseInt(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_NUM_RETRIES.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_NUM_RETRIES.defaultValue())))
+ .withClientRetryWaitTimeInMillis(Long.parseLong(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS.defaultValue())))
+ .withLockWaitTimeInMillis(Long.valueOf(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_WAIT_TIMEOUT_MS.key(),
+ String.valueOf(HoodieLockConfig.LOCK_ACQUIRE_WAIT_TIMEOUT_MS.defaultValue()))))
+ .withNumRetries(Integer.parseInt(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_NUM_RETRIES.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_NUM_RETRIES.defaultValue())))
+ .withRetryWaitTimeInMillis(Long.parseLong(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS.defaultValue())))
+ .withRetryMaxWaitTimeInMillis(Long.parseLong(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS.defaultValue())))
+ .withHeartbeatIntervalInMillis(Long.valueOf(props.getString(
+ HoodieLockConfig.LOCK_HEARTBEAT_INTERVAL_MS.key(),
+ String.valueOf(HoodieLockConfig.LOCK_HEARTBEAT_INTERVAL_MS.defaultValue()))));
+
+ String lockProviderClass = writeConfig.getLockProviderClass();
+ if (lockProviderClass == null) {
+ return lockConfigBuilder.build();
+ }
+
+ Properties providerProp = new Properties();
+ providerProp.setProperty(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key(), lockProviderClass);
+ lockConfigBuilder.fromProperties(providerProp);
+
+ if (ZookeeperBasedLockProvider.class.getName().equals(lockProviderClass)) {
+ if (props.containsKey(HoodieLockConfig.ZK_CONNECT_URL.key())) {
+ lockConfigBuilder.withZkQuorum(props.getString(HoodieLockConfig.ZK_CONNECT_URL.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.ZK_BASE_PATH.key())) {
+ lockConfigBuilder.withZkBasePath(props.getString(HoodieLockConfig.ZK_BASE_PATH.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.ZK_LOCK_KEY.key())) {
+ lockConfigBuilder.withZkLockKey(props.getString(HoodieLockConfig.ZK_LOCK_KEY.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.ZK_PORT.key())) {
+ lockConfigBuilder.withZkPort(props.getString(HoodieLockConfig.ZK_PORT.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.ZK_SESSION_TIMEOUT_MS.key())) {
+ lockConfigBuilder.withZkSessionTimeoutInMs(
+ Long.valueOf(props.getString(HoodieLockConfig.ZK_SESSION_TIMEOUT_MS.key())));
+ }
+ if (props.containsKey(HoodieLockConfig.ZK_CONNECTION_TIMEOUT_MS.key())) {
+ lockConfigBuilder.withZkConnectionTimeoutInMs(
+ Long.valueOf(props.getString(HoodieLockConfig.ZK_CONNECTION_TIMEOUT_MS.key())));
+ }
+ } else if (FileSystemBasedLockProvider.class.getName().equals(lockProviderClass)) {
+ if (props.containsKey(HoodieLockConfig.FILESYSTEM_LOCK_PATH.key())) {
+ lockConfigBuilder.withFileSystemLockPath(props.getString(HoodieLockConfig.FILESYSTEM_LOCK_PATH.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.FILESYSTEM_LOCK_EXPIRE.key())) {
+ lockConfigBuilder.withFileSystemLockExpire(
+ Integer.parseInt(props.getString(HoodieLockConfig.FILESYSTEM_LOCK_EXPIRE.key())));
+ }
+ } else if (lockProviderClass.contains("HiveMetastoreBasedLockProvider")) {
+ if (props.containsKey(HoodieLockConfig.HIVE_DATABASE_NAME.key())) {
+ lockConfigBuilder.withHiveDatabaseName(props.getString(HoodieLockConfig.HIVE_DATABASE_NAME.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.HIVE_TABLE_NAME.key())) {
+ lockConfigBuilder.withHiveTableName(props.getString(HoodieLockConfig.HIVE_TABLE_NAME.key()));
+ }
+ if (props.containsKey(HoodieLockConfig.HIVE_METASTORE_URI.key())) {
+ lockConfigBuilder.withHiveMetastoreURIs(props.getString(HoodieLockConfig.HIVE_METASTORE_URI.key()));
+ }
+ } else {
+ // For any custom lock provider, pass through all lock-prefixed properties
+ // so provider-specific configs are preserved.
Review Comment:
can we simplify and do this for all? Is there any case, where this may not work?
btw, how are alternate keys if any are accounted here?
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java:
##########
@@ -362,6 +387,99 @@ public static HoodieWriteConfig createMetadataWriteConfig(
return metadataWriteConfig;
}
+ /**
+ * Build the lock config for the metadata table by extracting lock-specific properties from the
+ * data table's write config. This avoids copying all properties (which would overwrite MDT-specific
+ * settings like base path and auto-clean).
+ */
+ private static HoodieLockConfig buildMetadataLockConfig(HoodieWriteConfig writeConfig) {
+ TypedProperties props = writeConfig.getProps();
+ HoodieLockConfig.Builder lockConfigBuilder = HoodieLockConfig.newBuilder()
+ .withClientNumRetries(Integer.parseInt(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_NUM_RETRIES.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_NUM_RETRIES.defaultValue())))
+ .withClientRetryWaitTimeInMillis(Long.parseLong(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS.defaultValue())))
+ .withLockWaitTimeInMillis(Long.valueOf(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_WAIT_TIMEOUT_MS.key(),
+ String.valueOf(HoodieLockConfig.LOCK_ACQUIRE_WAIT_TIMEOUT_MS.defaultValue()))))
+ .withNumRetries(Integer.parseInt(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_NUM_RETRIES.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_NUM_RETRIES.defaultValue())))
+ .withRetryWaitTimeInMillis(Long.parseLong(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS.defaultValue())))
+ .withRetryMaxWaitTimeInMillis(Long.parseLong(props.getString(
+ HoodieLockConfig.LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS.key(),
+ HoodieLockConfig.LOCK_ACQUIRE_RETRY_MAX_WAIT_TIME_IN_MILLIS.defaultValue())))
+ .withHeartbeatIntervalInMillis(Long.valueOf(props.getString(
+ HoodieLockConfig.LOCK_HEARTBEAT_INTERVAL_MS.key(),
+ String.valueOf(HoodieLockConfig.LOCK_HEARTBEAT_INTERVAL_MS.defaultValue()))));
+
+ String lockProviderClass = writeConfig.getLockProviderClass();
+ if (lockProviderClass == null) {
Review Comment:
shouldn't we throw here?
how come data table's write config does not have any lock provider class set?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org