nsivabalan commented on code in PR #19033:
URL: https://github.com/apache/hudi/pull/19033#discussion_r3762993800
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HiveQueryDDLExecutor.java:
##########
@@ -209,21 +217,84 @@ 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);
+ // Resolved here, on the calling thread, rather than inside the workers:
this is the
+ // only sync path that would otherwise call a user-supplied
PartitionValueExtractor
+ // from several threads at once. Extractors are pluggable and not
required to be
+ // thread-safe, and a garbled clause would drop the wrong partition. It
also halves
+ // the extractor calls, since partitionExists and the drop clause share
the values.
+ List<PartitionToDrop> resolved = new
ArrayList<>(partitionsToDrop.size());
+ for (String partition : partitionsToDrop) {
+ List<String> values =
partitionValueExtractor.extractPartitionValuesInPath(partition);
+ resolved.add(new PartitionToDrop(partition, values,
+ HivePartitionUtil.getPartitionClauseForDrop(values, config)));
}
+
+ int batchSyncPartitionNum =
config.getIntOrDefault(HIVE_BATCH_SYNC_PARTITION_NUM);
+ List<List<PartitionToDrop>> batches = CollectionUtils.batches(resolved,
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<PartitionToDrop>>
batches) throws Exception {
+ if (!metaStoreClientPool.isPresent()) {
+ for (List<PartitionToDrop> batch : batches) {
+ applyDropBatch(metaStoreClient, tableName, batch);
+ }
+ return;
+ }
+ HiveMetaStoreClientPool pool = metaStoreClientPool.get();
+ pool.awaitAll(
+ pool.dispatchAll(batches, (client, batch) -> applyDropBatch(client,
tableName, batch)),
+ "drop partition");
+ }
+
+ private void applyDropBatch(IMetaStoreClient client, String tableName,
List<PartitionToDrop> batch) throws Exception {
+ int dropped = 0;
+ for (PartitionToDrop dropPartition : batch) {
+ if (HivePartitionUtil.partitionExists(client, tableName,
dropPartition.path,
+ dropPartition.values, config)) {
+ client.dropPartition(databaseName, tableName, dropPartition.clause,
false);
+ dropped++;
+ }
+ // Per-partition detail stays at debug: a batch can hold thousands of
partitions
+ // and N workers log concurrently, so INFO carries the per-batch summary
instead.
+ log.debug("Dropped partition {} on {}", dropPartition.path, tableName);
+ }
+ log.info("Dropped {} of {} partitions in batch on {}", dropped,
batch.size(), tableName);
+ }
+
+ /**
+ * A partition to drop with its extractor-derived values already resolved,
so worker
+ * threads never touch the shared {@link PartitionValueExtractor}.
Immutable: the
+ * {@code values} list is wrapped unmodifiable at construction.
+ */
+ private static final class PartitionToDrop {
+ private final String path;
+ private final List<String> values;
+ private final String clause;
+
+ private PartitionToDrop(String path, List<String> values, String clause) {
+ this.path = path;
+ this.values = Collections.unmodifiableList(values);
Review Comment:
Good catch — fixed in 749d6719, using exactly the form you suggested:
```java
this.values = Collections.unmodifiableList(new ArrayList<>(values));
```
This one is squarely my miss. The whole point of that change was to make the
DROP path safe against a *stateful* extractor, and I guarded the concurrency
angle while leaving the aliasing one wide open — `unmodifiableList` protects
the caller from mutating the list, not the list from being mutated underneath
us.
Reproduced before fixing, with an extractor that clears and refills one list
per call across four partitions:
```
each partition must be checked with the values extracted for it
==> expected: <[2026-08-01, 2026-08-02, 2026-08-03, 2026-08-04]>
but was: <[2026-08-04]>
```
All four collapsed onto the final extraction, so `partitionExists` checked
`2026-08-04` four times — three valid drops silently skipped, exactly as you
described. Covered by `extractorReusingOneBufferStillYieldsPerPartitionValues`.
I also swept the other collections crossing a boundary in this PR while I
was here: `HiveDriverPool.workers`, `HiveMetaStoreClientPool.all`, and
`ParallelDispatch.futures` all already copy at construction (`new
ArrayList<>(...)`). `PartitionToDrop` was the only one wrapping without copying.
--
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]
