hudi-agent commented on code in PR #19761:
URL: https://github.com/apache/hudi/pull/19761#discussion_r3867806462
##########
hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java:
##########
@@ -377,34 +380,62 @@ public void updatePartitionsToTable(String tableName,
List<String> changedPartit
return;
}
Table table = getTable(awsGlue, databaseName, tableName);
- parallelizeChange(changedPartitions, this.changeParallelism, partitions
-> this.updatePartitionsToTableInternal(table, partitions),
MAX_PARTITIONS_PER_CHANGE_REQUEST);
+ parallelizeChange(changedPartitions, this.changeParallelism,
+ batch -> this.updatePartitionsInternal(table, () ->
partitionsFromStoragePaths(batch), UPDATE_PARTITIONS),
MAX_PARTITIONS_PER_CHANGE_REQUEST);
} finally {
log.info("Updated {} partitions to table {} in {} ms",
changedPartitions.size(), tableId(this.databaseName, tableName),
timer.endTimer());
}
}
- private void updatePartitionsToTableInternal(Table table, List<String>
changedPartitions) {
+ /** Builds Partitions whose location is derived from the storage path, not
read back from the catalog. */
+ private List<Partition> partitionsFromStoragePaths(List<String>
storagePartitionPaths) {
+ return storagePartitionPaths.stream()
+ .map(p -> new Partition(
+ partitionValueExtractor.extractPartitionValuesInPath(p),
+ FSUtils.constructAbsolutePath(s3aToS3(getBasePath()),
p).toString()))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Propagates the table's columns onto every partition, reusing each
partition's recorded location:
+ * a location derived from partition values can miss the real layout and
point at a prefix with no data.
+ */
+ private void cascadeColumnsToPartitions(String tableName, List<Partition>
partitions) {
+ HoodieTimer timer = HoodieTimer.start();
try {
+ if (partitions.isEmpty()) {
+ log.info("No partitions to cascade columns to for {}",
tableId(this.databaseName, tableName));
+ return;
+ }
+ Table table = getTable(awsGlue, databaseName, tableName);
+ parallelizeChange(partitions, this.changeParallelism,
+ batch -> this.updatePartitionsInternal(table, () -> batch,
CASCADE_COLUMNS_TO_PARTITIONS), MAX_PARTITIONS_PER_CHANGE_REQUEST);
+ } finally {
+ log.info("Cascaded columns to {} partitions of table {} in {} ms",
partitions.size(), tableId(this.databaseName, tableName), timer.endTimer());
+ }
+ }
+
+ /** Partitions are supplied lazily so a failure deriving them is wrapped
here rather than escaping unwrapped. */
+ private void updatePartitionsInternal(Table table, Supplier<List<Partition>>
partitionsSupplier, String context) {
Review Comment:
🤖 nit: `context` is a bit too broad here — since this parameter is really an
action phrase spliced into error messages (`"Fail to " + context + " " +
tableId(...)`), something like `operationLabel` or `actionDescription` would
make the intent clearer at each call site without needing to read the body.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
