danny0405 commented on code in PR #19033:
URL: https://github.com/apache/hudi/pull/19033#discussion_r3662480717
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java:
##########
@@ -227,6 +241,22 @@ private IMetaStoreClient
createMetaStoreClient(HiveSyncConfig config) {
}
}
+ private Option<IMetaStoreClientPool>
maybeBuildPartitionClientPool(HiveSyncConfig config) {
+ if (!config.getBooleanOrDefault(HIVE_SYNC_BATCHING_ENABLED)) {
+ return Option.empty();
+ }
+ if (config.getBooleanOrDefault(HIVE_SYNC_USE_SPARK_CATALOG)) {
+ // The Spark catalog client is constructed via reflection against a
Spark-side
+ // class and isn't compatible with the direct RetryingMetaStoreClient
pool path.
+ // Fall back to single-client sequential behavior rather than failing
the sync.
+ log.warn("hive_sync.batching.enabled=true is not supported with
use_spark_catalog=true; "
+ + "falling back to sequential partition sync.");
+ return Option.empty();
+ }
+ int size = config.getIntOrDefault(HIVE_SYNC_BATCHING_THREADS);
+ return Option.of(new IMetaStoreClientPool(config, size));
Review Comment:
**[P2] Update the public batching config contract for DROP**
`HIVE_SYNC_BATCHING_ENABLED` in `HiveSyncConfigHolder` still documents that
"DROP remains serial," but this pool makes DROP parallel under that exact flag.
That text is the user-facing configuration contract, so it currently tells
operators the opposite of the new behavior. Please update it to cover the
Thrift-client DROP fan-out (including the Spark-catalog fallback) as part of
this change.
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HiveQueryDDLExecutor.java:
##########
@@ -209,21 +219,96 @@ public void dropPartitionsToTable(String tableName,
List<String> partitionsToDro
log.info("Drop partitions {} on {}", partitionsToDrop.size(), tableName);
try {
- for (String dropPartition : partitionsToDrop) {
- if (HivePartitionUtil.partitionExists(metaStoreClient, tableName,
dropPartition, partitionValueExtractor,
- config)) {
- String partitionClause =
- HivePartitionUtil.getPartitionClauseForDrop(dropPartition,
partitionValueExtractor, config);
- metaStoreClient.dropPartition(databaseName, tableName,
partitionClause, false);
- }
- log.info("Drop partition {} on {}", dropPartition, tableName);
- }
+ int batchSyncPartitionNum =
config.getIntOrDefault(HIVE_BATCH_SYNC_PARTITION_NUM);
+ List<List<String>> batches = CollectionUtils.batches(partitionsToDrop,
batchSyncPartitionNum);
+ runDropBatches(tableName, batches);
} catch (Exception e) {
log.error("{} drop partition failed", tableId(databaseName, tableName),
e);
throw new HoodieHiveSyncException(tableId(databaseName, tableName) + "
drop partition failed", e);
}
}
+ /**
+ * Drops partitions one batch at a time. When {@link #metaStoreClientPool}
is present,
+ * batches fan out across the pool's worker threads (each borrowing an
independent
+ * IMetaStoreClient); otherwise batches are dispatched sequentially against
the
+ * session client. Hive has no batch-drop primitive that matches
dropPartition's
+ * semantics, so each worker still iterates its chunk one partition at a
time — the
+ * win is fanning chunks across independent Thrift clients.
+ *
+ * <p>First-error semantics match {@code HiveDriverPool.awaitAll}: the first
failure is
+ * rethrown, not-yet-started batches are cancelled, and later failures are
logged at WARN.
+ */
+ private void runDropBatches(String tableName, List<List<String>> batches)
throws Exception {
+ if (!metaStoreClientPool.isPresent()) {
+ for (List<String> batch : batches) {
+ applyDropBatch(metaStoreClient, tableName, batch);
+ }
+ return;
+ }
+ IMetaStoreClientPool pool = metaStoreClientPool.get();
+ List<Future<Void>> futures = new ArrayList<>(batches.size());
+ for (List<String> batch : batches) {
+ futures.add(pool.executor().submit(() ->
+ pool.run(poolClient -> {
+ applyDropBatch(poolClient, tableName, batch);
+ return null;
+ })
+ ));
+ }
+
+ Exception firstError = null;
+ int cancelled = 0;
+ for (Future<Void> f : futures) {
+ // Once a batch has failed the sync is going to abort anyway, so stop
handing
+ // more DROPs to the metastore. mayInterruptIfRunning=false mirrors
+ // HiveDriverPool.awaitAll: a batch already mid-flight runs to
completion rather
+ // than leaving the partition list half-dropped at an arbitrary point.
+ if (firstError != null && f.cancel(false)) {
+ cancelled++;
+ continue;
+ }
+ try {
+ f.get();
Review Comment:
**[P2] Stop queued DROP batches as soon as a worker fails**
This waits on futures in submission order. If a later batch fails quickly
while an earlier batch is slow, this thread remains blocked on the earlier
`f.get()`, and the shared executor can keep starting every queued batch because
the tasks do not observe an abort flag. By the time `firstError` is set, most
or all "not-yet-started" DROPs may already have run, so the advertised
abort-on-first-error behavior is not achieved. Please mirror
`HiveDriverPool.Dispatch` (shared task-side abort plus completion-order
notification), or use a completion service/abort flag so pending batches are
stopped immediately after any worker fails.
--
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]
