feat(hive-sync): parallelize DROP partitions in HiveQL sync mode #19033
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,12 +18,14 @@ | |
|
|
||
| package org.apache.hudi.hive.ddl; | ||
|
|
||
| import org.apache.hudi.common.util.CollectionUtils; | ||
| import org.apache.hudi.common.util.HoodieTimer; | ||
| import org.apache.hudi.common.util.Option; | ||
| import org.apache.hudi.hive.HiveSyncConfig; | ||
| import org.apache.hudi.hive.HoodieHiveSyncException; | ||
| import org.apache.hudi.hive.util.HiveDriverPool; | ||
| import org.apache.hudi.hive.util.HivePartitionUtil; | ||
| import org.apache.hudi.hive.util.IMetaStoreClientPool; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.hadoop.hive.metastore.IMetaStoreClient; | ||
|
|
@@ -41,6 +43,8 @@ | |
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.concurrent.CancellationException; | ||
| import java.util.concurrent.Future; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_BATCH_SYNC_PARTITION_NUM; | ||
|
|
@@ -59,16 +63,22 @@ public class HiveQueryDDLExecutor extends QueryBasedDDLExecutor { | |
| // (createTable, schema evolution, single-statement runSQL callers) always uses the | ||
| // session `hiveDriver` above. See HiveDriverPool javadoc. | ||
| private final Option<HiveDriverPool> driverPool; | ||
| // When present, dropPartitionsToTable fans batches across this Thrift client pool. | ||
| // Owned by HoodieHiveSyncClient; close() is delegated through there. See | ||
| // IMetaStoreClientPool javadoc for the usage contract (partition-row ops only). | ||
| private final Option<IMetaStoreClientPool> metaStoreClientPool; | ||
|
|
||
| public HiveQueryDDLExecutor(HiveSyncConfig config, IMetaStoreClient metaStoreClient) { | ||
| this(config, metaStoreClient, Option.empty(), Option.empty()); | ||
| } | ||
|
|
||
| public HiveQueryDDLExecutor(HiveSyncConfig config, IMetaStoreClient metaStoreClient, | ||
| Option<HiveDriverPool> driverPool, | ||
| Option<IMetaStoreClientPool> metaStoreClientPool) { | ||
| super(config); | ||
| this.metaStoreClient = metaStoreClient; | ||
| this.driverPool = driverPool; | ||
| this.metaStoreClientPool = metaStoreClientPool; | ||
| try { | ||
| this.sessionState = new SessionState(config.getHiveConf(), | ||
| UserGroupInformation.getCurrentUser().getShortUserName()); | ||
|
|
@@ -209,21 +219,96 @@ public void dropPartitionsToTable(String tableName, List<String> partitionsToDro | |
|
|
||
| log.info("Drop partitions {} on {}", partitionsToDrop.size(), tableName); | ||
| try { | ||
| 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 { | ||
|
Contributor
There was a problem hiding this comment. 🤖 nit: the parallel-path error-collection loop inside - AI-generated; verify before applying. React 👍/👎 to flag quality.
Contributor
Author
There was a problem hiding this comment. Good suggestion — once #18984 merges and this branch rebases, HiveDriverPool.awaitAll(List<Future<?>>) will be available; will look at whether IMetaStoreClientPool's runDropBatches error-collection loop here can delegate to an analogous awaitAll on IMetaStoreClientPool at that point, to de-duplicate the two nearly-identical loops. |
||
| 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(); | ||
|
Contributor
There was a problem hiding this 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 |
||
| } catch (CancellationException ce) { | ||
| cancelled++; | ||
| } catch (InterruptedException ie) { | ||
| Thread.currentThread().interrupt(); | ||
| if (firstError == null) { | ||
| firstError = ie; | ||
| } | ||
| } catch (Exception e) { | ||
| if (firstError == null) { | ||
| firstError = e; | ||
| } else { | ||
| log.warn("Additional drop batch failed on {} (suppressed in favor of first error)", tableName, e); | ||
| } | ||
| } | ||
| } | ||
| if (firstError != null) { | ||
| log.error("Drop partition dispatch on {} aborted after first failure ({} batches cancelled)", | ||
| tableName, cancelled); | ||
| throw firstError; | ||
| } | ||
| } | ||
|
|
||
| private void applyDropBatch(IMetaStoreClient poolClient, String tableName, List<String> batch) throws Exception { | ||
| int dropped = 0; | ||
|
Contributor
There was a problem hiding this comment. 🤖 nit: the parameter is named |
||
| for (String dropPartition : batch) { | ||
| if (HivePartitionUtil.partitionExists(poolClient, tableName, dropPartition, | ||
| partitionValueExtractor, config)) { | ||
| String partitionClause = | ||
| HivePartitionUtil.getPartitionClauseForDrop(dropPartition, partitionValueExtractor, config); | ||
| poolClient.dropPartition(databaseName, tableName, partitionClause, 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, tableName); | ||
| } | ||
| log.info("Dropped {} of {} partitions in batch on {}", dropped, batch.size(), tableName); | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| // Close the pool first so the worker threads stop dispatching against their | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🤖 nit: the config key names in this warning (
hive_sync.batching.enabled,use_spark_catalog) don't match the actual property keys an operator would search for in logs. Could you use the real constant's.key()value, e.g.HIVE_SYNC_BATCHING_ENABLED.key() + "=true is not supported with " + HIVE_SYNC_USE_SPARK_CATALOG.key() + "=true"?