shangxinli commented on code in PR #18983:
URL: https://github.com/apache/hudi/pull/18983#discussion_r4433916095
##########
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.
Review Comment:
This javadoc claim is wrong for the pool path: all batches are submitted up
front and the loop below just drains futures in order, so batch N+1 keeps
running even after batch N already threw.
##########
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);
+ return null;
+ })
+ ));
+ }
+ Exception firstError = null;
+ for (Future<Void> f : futures) {
+ try {
+ f.get();
+ } catch (Exception e) {
+ if (firstError == null) {
+ firstError = e;
Review Comment:
firstError is the raw ExecutionException from f.get(), never unwrapped, so
the HoodieHiveSyncException thrown by the caller ends up with an
ExecutionException as its immediate cause instead of the original HMS exception.
--
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]
