danny0405 commented on code in PR #19033: URL: https://github.com/apache/hudi/pull/19033#discussion_r3739710143
########## hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/util/ParallelDispatch.java: ########## @@ -0,0 +1,272 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.hive.util; + +import org.apache.hudi.common.util.VisibleForTesting; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Handle to one fan-out batch of partition work: the submitted futures plus the shared + * abort flag the tasks consult before running. + * + * <p>The abort flag exists because waiting on futures in <i>submission</i> order is not + * enough to stop queued work. If a later task fails quickly while an earlier one is slow, + * the awaiting thread is still parked on the earlier {@code Future.get()}, and the + * executor happily keeps starting every queued task in the meantime. By the time the + * failure is observed, most of the "not-yet-started" work has already run. + * + * <p>Two mechanisms fix that, and both are needed: + * <ul> + * <li>a {@link CountDownLatch} tripped by the <i>first</i> abort, so the awaiting + * thread wakes on failure rather than on its turn in submission order;</li> + * <li>a task-side {@link #aborted()} check on entry, because cancelling from the + * awaiting thread is inherently late — a worker can pull its next task off the + * queue at any moment.</li> + * </ul> + * + * <p>The failing task also records its own throwable as it aborts, so the reported root + * cause is the failure that stopped the batch. Picking the error by scanning futures in + * submission order would instead surface whichever failure happens to sit earliest in the + * list, which is not necessarily the one that aborted the run. + * + * <p>Shared by {@link HiveDriverPool} (Hive {@code Driver} statements) and + * {@link HiveMetaStoreClientPool} (Thrift {@code dropPartition} batches), which fan out over + * different execution models but need identical abort-on-first-error semantics. + */ +public final class ParallelDispatch { + + private final List<Future<?>> futures; + private final int total; + private final AtomicInteger settled = new AtomicInteger(0); + private final AtomicBoolean aborted = new AtomicBoolean(false); + private final AtomicReference<Throwable> abortCause = new AtomicReference<>(); + private final CountDownLatch done = new CountDownLatch(1); + private volatile boolean sealed; + + ParallelDispatch(int total) { + this.total = total; + this.futures = new ArrayList<>(total); + } + + void add(Future<?> future) { + futures.add(future); + } + + // Called once submission finishes. A task that settles before the last submit + // would otherwise see settled < total and never trip the latch, so re-check here. + void sealed() { + sealed = true; + signalIfComplete(); + } + + boolean aborted() { + return aborted.get(); + } + + void abort() { + aborted.set(true); + done.countDown(); + } + + // Records the failure that triggered the abort, first writer wins. Selecting the error + // by walking futures in submission order instead would report whichever failure sits + // earliest in the list, not the one that actually stopped the batch: a fast failure at + // index 1 aborts the run, and a slow index 0 that fails later would take its place. + void abort(Throwable cause) { + abortCause.compareAndSet(null, cause); + abort(); + } + + void taskSettled() { + settled.incrementAndGet(); + signalIfComplete(); + } + + private void signalIfComplete() { + if (sealed && settled.get() >= total) { + done.countDown(); + } + } + + void awaitSettledOrAborted() { + if (total == 0) { + return; + } + try { + done.await(); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + aborted.set(true); + } Review Comment: **[P1] Propagate interruption as a failed dispatch** If the awaiting thread is interrupted here, this restores the interrupt flag and sets `aborted`, but never records the `InterruptedException` in `abortCause` or otherwise returns it to `awaitOutcome()`. The next `cancelPending()` can mark every future cancelled, so the future drain sees only `CancellationException`; `firstError` remains null and both pool `awaitAll()` methods report success. On the DROP path, `HiveSyncTool` can then advance the last-synced commit marker even though some partition operations were cancelled or are still running. Please preserve the interrupt as the dispatch failure (and ensure the caller cannot observe success before in-flight work settles). A regression test can block a dispatched action, interrupt the thread in `awaitAll`, and assert that `awaitAll` throws rather than returning normally. ########## 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: **[P2] Defensively copy the extractor result** `Collections.unmodifiableList(values)` only wraps the supplied list; it does not make a snapshot. `PartitionValueExtractor` is pluggable and its interface does not require returning a fresh or immutable list, so a custom extractor that reuses a mutable buffer leaves every `PartitionToDrop.values` pointing at the values from the final extraction. The clauses were materialized correctly, but `partitionExists()` then checks the wrong partition for earlier entries, which can skip valid drops or cause a spurious failure. Please copy before wrapping, for example `Collections.unmodifiableList(new ArrayList<>(values))`, and cover this with an extractor that deliberately reuses one list. -- 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]
