nsivabalan commented on code in PR #19033:
URL: https://github.com/apache/hudi/pull/19033#discussion_r3643359984
##########
hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/ddl/HMSDDLExecutor.java:
##########
@@ -291,21 +312,71 @@ public void close() {
private void registerAlterPartitionEvent(String tableName, List<String>
alteredPartitions) {
try {
+ // Read the StorageDescriptor once on the session client; each worker
deep-copies
+ // it per partition (alter_partitions semantics today).
StorageDescriptor sd = client.getTable(databaseName, tableName).getSd();
- List<Partition> partitionList = alteredPartitions.stream().map(partition
-> {
- Path partitionPath =
HadoopFSUtils.constructAbsolutePathInHadoopPath(syncConfig.getString(META_SYNC_BASE_PATH),
partition);
- String partitionScheme = partitionPath.toUri().getScheme();
- String fullPartitionPath =
StorageSchemes.HDFS.getScheme().equals(partitionScheme)
- ?
HadoopFSUtils.getDFSFullPartitionPath(syncConfig.getHadoopFileSystem(),
partitionPath) : partitionPath.toString();
- List<String> partitionValues =
partitionValueExtractor.extractPartitionValuesInPath(partition);
- StorageDescriptor partitionSd = sd.deepCopy();
- partitionSd.setLocation(fullPartitionPath);
- return new Partition(partitionValues, databaseName, tableName, 0, 0,
partitionSd, null);
- }).collect(Collectors.toList());
- client.alter_partitions(databaseName, tableName, partitionList, null);
- } catch (TException e) {
+ int batchSyncPartitionNum =
syncConfig.getIntOrDefault(HIVE_BATCH_SYNC_PARTITION_NUM);
+ List<List<String>> batches = CollectionUtils.batches(alteredPartitions,
batchSyncPartitionNum);
+ runBatches("alter", tableName, batches, (poolClient, batch) -> {
+ List<Partition> partitionList = batch.stream().map(partition -> {
+ Path partitionPath =
HadoopFSUtils.constructAbsolutePathInHadoopPath(syncConfig.getString(META_SYNC_BASE_PATH),
partition);
+ String partitionScheme = partitionPath.toUri().getScheme();
+ String fullPartitionPath =
StorageSchemes.HDFS.getScheme().equals(partitionScheme)
+ ?
HadoopFSUtils.getDFSFullPartitionPath(syncConfig.getHadoopFileSystem(),
partitionPath) : partitionPath.toString();
+ List<String> partitionValues =
partitionValueExtractor.extractPartitionValuesInPath(partition);
+ StorageDescriptor partitionSd = sd.deepCopy();
+ partitionSd.setLocation(fullPartitionPath);
+ return new Partition(partitionValues, databaseName, tableName, 0, 0,
partitionSd, null);
+ }).collect(Collectors.toList());
+ poolClient.alter_partitions(databaseName, tableName, partitionList,
null);
+ });
+ } catch (Exception e) {
log.error("{}.{} update partition failed", databaseName, tableName, e);
throw new HoodieHiveSyncException(databaseName + "." + tableName + "
update partition failed", e);
}
}
+
+ /**
+ * Dispatches partition batches either in parallel against the client pool
(if
+ * configured) or sequentially against the session client. The sequential
path
+ * preserves the exact behavior we had before the pool existed, including
failure
+ * semantics where batch N+1 never runs if batch N throws.
+ */
+ @FunctionalInterface
+ private interface BatchAction {
+ void apply(IMetaStoreClient client, List<String> batch) throws Exception;
+ }
+
+ private void runBatches(String opName, String tableName, List<List<String>>
batches, BatchAction action) throws Exception {
+ if (partitionClientPool == null) {
+ for (List<String> batch : batches) {
+ action.apply(client, batch);
+ }
+ return;
+ }
+ List<Future<Void>> futures = new ArrayList<>(batches.size());
+ for (List<String> batch : batches) {
+ futures.add(partitionClientPool.executor().submit(() ->
+ partitionClientPool.run(poolClient -> {
+ action.apply(poolClient, batch);
Review Comment:
This references HMSDDLExecutor.runBatches / registerAlterPartitionEvent(with
pool), which isn't part of this PR's diff — HMSDDLExecutor is untouched here.
That code belonged to an HMS-parallelization commit that was on this branch at
an earlier point but has since been dropped (that work will land as its own
separate PR). Resolving as not applicable to the current diff; will revisit the
awaitAll-reuse suggestion when that HMS PR is opened.
--
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]
