nsivabalan commented on code in PR #19033:
URL: https://github.com/apache/hudi/pull/19033#discussion_r3737119697
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HiveQueryDDLExecutor.java:
##########
@@ -209,21 +217,57 @@ 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 come from {@link ParallelDispatch}, shared with
+ * {@code HiveDriverPool}: the first failure is rethrown, batches that have
not started
+ * are stopped via the task-side abort flag, 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();
+ pool.awaitAll(
+ pool.dispatchAll(batches, (client, batch) -> applyDropBatch(client,
tableName, batch)),
+ "drop partition");
+ }
+
+ private void applyDropBatch(IMetaStoreClient client, String tableName,
List<String> batch) throws Exception {
+ int dropped = 0;
+ for (String dropPartition : batch) {
+ if (HivePartitionUtil.partitionExists(client, tableName, dropPartition,
Review Comment:
Filed as a nit, but I agree with your read that it is more than that — this
is the data-loss-shaped one. Went with your second suggestion (extract on the
calling thread before fan-out) rather than documenting the requirement, in
3a1694a.
Confirmed it was real before fixing it. A test that records the thread of
every `extractPartitionValuesInPath` call, with a 4-client pool and 16
partitions, fails on the old code with:
```
expected: <[main]>
but was: <[hudi-hive-sync-pool-1-1, hudi-hive-sync-pool-1-2,
hudi-hive-sync-pool-1-3, hudi-hive-sync-pool-1-4, main]>
```
All four workers were in the shared extractor concurrently, exactly as you
described.
The fix resolves each partition once on the calling thread into an immutable
carrier (path + values + drop clause) and fans out only that. Three side
benefits beyond removing the concurrency assumption:
- **Halves the extractor calls.** `partitionExists` and
`getPartitionClauseForDrop` were each extracting the same partition separately;
they now share one result.
- **Fails before the first drop.** A malformed partition previously threw
partway through the batch, after some partitions had already been dropped. It
now fails ahead of any drop — same `HoodieHiveSyncException`, just earlier,
which matters for a destructive operation.
- **`HMSDDLExecutor` is untouched.** `HivePartitionUtil` gained overloads
taking pre-extracted values and the existing signatures delegate to them, so
the sequential HMS path keeps its current behaviour.
New test:
`TestDropPartitionExtractorThreading#extractorIsNeverInvokedFromAPoolWorker`.
It lives in `hive.util` rather than beside the executor because it needs the
package-private `IMetaStoreClientPool` constructor that accepts pre-built
clients. It also asserts the drops actually reached the pool, so it cannot pass
by silently never fanning out.
--
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]
