nsivabalan commented on code in PR #19033:
URL: https://github.com/apache/hudi/pull/19033#discussion_r3669746061
##########
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:
Good catch — you are right, and the abort was not merely weaker than
advertised, it was doing nothing. I wrote a unit test against the previous code
and every queued batch still ran after the first failure:
```
expected: <[FAIL]> but was: <[FAIL, AFTER_A, AFTER_B]>
```
The loop waited on futures in submission order, so a fast failure on a later
batch went unobserved while the awaiting thread was parked on an earlier slow
one, and the executor kept starting queued tasks the whole time.
`Future.cancel` from the awaiting thread is always too late once a worker has
dequeued its next task.
Fixed by taking your suggestion and extracting `HiveDriverPool.Dispatch`
into a shared `ParallelDispatch` rather than reimplementing the pattern a
second time — so both mechanisms that make it work are now shared: the latch
tripped by the first abort (awaiting thread wakes on failure, not on its turn
in submission order) and the task-side abort check on entry.
`IMetaStoreClientPool` now has `dispatchAll`/`awaitAll` built on it, and
`runDropBatches` delegates instead of hand-rolling the loop — which also
resolves the duplicate-loop nit hudi-agent raised earlier. Net effect is -139
lines.
Covered by two regression tests in the new `TestIMetaStoreClientPool`:
`abortStopsQueuedBatchesAfterFirstFailure` and
`abortStopsQueuedBatchesWhenEarlierBatchIsSlow` (the latter pins exactly the
slow-earlier-future interleaving you described). Both fail against the previous
implementation. Pushed in `8721c9a`.
##########
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:
Agreed — that text was the pre-DROP contract and now states the opposite of
the behavior. Fixed in `6c81322`.
`HIVE_SYNC_BATCHING_ENABLED` now documents that DROP is parallelized too,
that it fans out over metastore (Thrift) clients rather than Hive Driver
workers (since it is issued as `dropPartition` calls rather than SQL), that
drops are split into `batch_num`-sized batches, and that it falls back to
sequential execution on the single session client when `use_spark_catalog=true`.
Also updated `batching.threads`, which the same value now sizes for both
pools — that was equally misleading since it only mentioned Driver workers. And
switched the spark-catalog fallback warning to emit real config keys via
`.key()` (hudi-agent nit on the same change), so operators can grep the
property they actually 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
