This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new fe82430e694a fix(sync): keep each partition's recorded location when
cascading Glue schema changes (#19761)
fe82430e694a is described below
commit fe82430e694ad00d1e405198054efb2a88aef56d
Author: niranjan-1408 <[email protected]>
AuthorDate: Thu Aug 27 17:12:52 2026 -0700
fix(sync): keep each partition's recorded location when cascading Glue
schema changes (#19761)
---
.../hudi/aws/sync/AWSGlueCatalogSyncClient.java | 67 ++++++++++++-------
.../hudi/aws/sync/TestAWSGlueSyncClient.java | 75 +++++++++++++++++++++-
2 files changed, 117 insertions(+), 25 deletions(-)
diff --git
a/hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java
b/hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java
index eb0a50ca0638..f587882dfdf8 100644
---
a/hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java
+++
b/hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java
@@ -103,6 +103,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Consumer;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
import static org.apache.hudi.common.fs.FSUtils.s3aToS3;
@@ -148,6 +149,8 @@ public class AWSGlueCatalogSyncClient extends
HoodieSyncClient {
*/
private static final String ENABLE_MDT_LISTING =
"hudi.metadata-listing-enabled";
private static final String GLUE_TABLE_ARN_FORMAT =
"arn:aws:glue:%s:%s:table/%s/%s";
+ private static final String UPDATE_PARTITIONS = "update partitions to";
+ private static final String CASCADE_COLUMNS_TO_PARTITIONS = "cascade columns
to partitions of";
private static final String GLUE_DATABASE_ARN_FORMAT =
"arn:aws:glue:%s:%s:database/%s";
@Getter
private final String databaseName;
@@ -377,34 +380,62 @@ public class AWSGlueCatalogSyncClient extends
HoodieSyncClient {
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) {
+ try {
+ List<Partition> partitions = partitionsSupplier.get();
StorageDescriptor sd = table.storageDescriptor();
- List<BatchUpdatePartitionRequestEntry> updatePartitionEntries =
changedPartitions.stream().map(partition -> {
- String fullPartitionPath =
FSUtils.constructAbsolutePath(s3aToS3(getBasePath()), partition).toString();
- List<String> partitionValues =
partitionValueExtractor.extractPartitionValuesInPath(partition);
- StorageDescriptor partitionSD = sd.copy(copySd ->
copySd.location(fullPartitionPath));
+ List<BatchUpdatePartitionRequestEntry> updatePartitionEntries =
partitions.stream().map(partition -> {
+ List<String> partitionValues = partition.getValues();
+ StorageDescriptor partitionSD = sd.copy(copySd ->
copySd.location(partition.getStorageLocation()));
PartitionInput partitionInput =
PartitionInput.builder().values(partitionValues).storageDescriptor(partitionSD).build();
return
BatchUpdatePartitionRequestEntry.builder().partitionInput(partitionInput).partitionValueList(partitionValues).build();
}).collect(Collectors.toList());
BatchUpdatePartitionRequest request =
BatchUpdatePartitionRequest.builder().catalogId(catalogId)
.databaseName(databaseName).tableName(table.name()).entries(updatePartitionEntries).build();
- CompletableFuture<BatchUpdatePartitionResponse> future =
awsGlue.batchUpdatePartition(request);
-
- BatchUpdatePartitionResponse response = future.get();
+ BatchUpdatePartitionResponse response =
awsGlue.batchUpdatePartition(request).get();
if (CollectionUtils.nonEmpty(response.errors())) {
- throw new HoodieGlueSyncException("Fail to update partitions to " +
tableId(databaseName, table.name())
+ throw new HoodieGlueSyncException("Fail to " + context + " " +
tableId(databaseName, table.name())
+ " with error(s): " + response.errors());
}
} catch (Exception e) {
- throw new HoodieGlueSyncException("Fail to update partitions to " +
tableId(databaseName, table.name()), e);
+ throw new HoodieGlueSyncException("Fail to " + context + " " +
tableId(databaseName, table.name()), e);
}
}
@@ -595,11 +626,7 @@ public class AWSGlueCatalogSyncClient extends
HoodieSyncClient {
// TODO: skip cascading when new fields in structs are added to the
schema in last position
boolean cascade =
config.getSplitStrings(META_SYNC_PARTITION_FIELDS).size() > 0 &&
!schemaDiff.getUpdateColumnTypes().isEmpty();
if (cascade) {
- log.info("Cascading column changes to partitions");
- List<String> allPartitions = getAllPartitions(tableName).stream()
- .map(partition -> getStringFromPartition(table.partitionKeys(),
partition.getValues()))
- .collect(Collectors.toList());
- updatePartitionsToTable(tableName, allPartitions);
+ cascadeColumnsToPartitions(tableName, getAllPartitions(tableName));
}
awsGlue.updateTable(request).get();
} catch (Exception e) {
@@ -607,14 +634,6 @@ public class AWSGlueCatalogSyncClient extends
HoodieSyncClient {
}
}
- private String getStringFromPartition(List<Column> partitionKeys,
List<String> values) {
- ArrayList<String> partitionValues = new ArrayList<>();
- for (int i = 0; i < partitionKeys.size(); i++) {
- partitionValues.add(String.format("%s=%s", partitionKeys.get(i).name(),
values.get(i)));
- }
- return partitionValues.stream().collect(Collectors.joining("/"));
- }
-
@Override
public void createOrReplaceTable(String tableName,
HoodieSchema storageSchema,
diff --git
a/hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java
b/hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java
index 8c3b24c0b677..6400198f0ac7 100644
--- a/hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java
+++ b/hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java
@@ -25,6 +25,8 @@ import org.apache.hudi.common.schema.HoodieSchema;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.config.GlueCatalogSyncClientConfig;
import org.apache.hudi.hive.HiveSyncConfig;
+import org.apache.hudi.hive.SchemaDifference;
+import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.sync.common.model.FieldSchema;
import org.apache.hudi.sync.common.model.Partition;
@@ -37,6 +39,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
@@ -48,6 +51,7 @@ import
software.amazon.awssdk.services.glue.model.BatchCreatePartitionResponse;
import software.amazon.awssdk.services.glue.model.BatchDeletePartitionRequest;
import software.amazon.awssdk.services.glue.model.BatchDeletePartitionResponse;
import software.amazon.awssdk.services.glue.model.BatchUpdatePartitionRequest;
+import
software.amazon.awssdk.services.glue.model.BatchUpdatePartitionRequestEntry;
import software.amazon.awssdk.services.glue.model.BatchUpdatePartitionResponse;
import software.amazon.awssdk.services.glue.model.Column;
import software.amazon.awssdk.services.glue.model.CreateDatabaseRequest;
@@ -99,6 +103,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
@@ -575,12 +580,19 @@ class TestAWSGlueSyncClient {
.thenReturn(CompletableFuture.completedFuture(gt));
BatchUpdatePartitionResponse ok =
BatchUpdatePartitionResponse.builder().errors(Collections.emptyList()).build();
-
when(mockAwsGlue.batchUpdatePartition(any(BatchUpdatePartitionRequest.class)))
+ ArgumentCaptor<BatchUpdatePartitionRequest> captor =
ArgumentCaptor.forClass(BatchUpdatePartitionRequest.class);
+ when(mockAwsGlue.batchUpdatePartition(captor.capture()))
.thenReturn(CompletableFuture.completedFuture(ok));
awsGlueSyncClient.updatePartitionsToTable(tableName, changed);
verify(mockAwsGlue).batchUpdatePartition(any(BatchUpdatePartitionRequest.class));
+ List<BatchUpdatePartitionRequestEntry> entries =
captor.getValue().entries();
+ assertEquals(1, entries.size());
+ String syncBasePath =
GlueTestUtil.getHiveSyncConfig().getString(META_SYNC_BASE_PATH);
+ assertEquals(new StoragePath(syncBasePath, "2025/05/20").toString(),
+ entries.get(0).partitionInput().storageDescriptor().location());
+ assertEquals(Collections.singletonList("2025-05-20"),
entries.get(0).partitionValueList());
}
@Test
@@ -909,4 +921,65 @@ class TestAWSGlueSyncClient {
.build();
return CompletableFuture.completedFuture(response);
}
+
+ @ParameterizedTest
+ @ValueSource(strings = {"2024-01-15", "datestr=2024-01-15", "2024/01/15"})
+ void
testUpdateTableSchema_cascadePreservesGlueRecordedPartitionLocation(String
partitionDir) {
+ String tableName = GlueTestUtil.TABLE_NAME;
+ String basePath =
GlueTestUtil.getHiveSyncConfig().getString(META_SYNC_BASE_PATH);
+ String partitionLocation = new StoragePath(basePath,
partitionDir).toString();
+
+ Table table = Table.builder()
+ .name(tableName)
+ .databaseName(GlueTestUtil.DB_NAME)
+ .storageDescriptor(StorageDescriptor.builder()
+ .location(basePath)
+ .columns(Column.builder().name("name").type("string").build())
+ .build())
+ .partitionKeys(Column.builder().name("datestr").type("string").build())
+ .build();
+ // the cascade re-reads the table after the schema update, so the second
read carries the new columns
+ Table updatedTable = table.toBuilder()
+ .storageDescriptor(table.storageDescriptor().toBuilder()
+ .columns(Column.builder().name("id").type("int").build(),
+ Column.builder().name("name").type("string").build())
+ .build())
+ .build();
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(table).build()))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(updatedTable).build()));
+ when(mockAwsGlue.updateTable(any(UpdateTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
+ software.amazon.awssdk.services.glue.model.Partition gluePartition =
+ software.amazon.awssdk.services.glue.model.Partition.builder()
+ .values("2024-01-15")
+
.storageDescriptor(StorageDescriptor.builder().location(partitionLocation).build())
+ .build();
+ when(mockAwsGlue.getPartitions(any(GetPartitionsRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+
GetPartitionsResponse.builder().partitions(gluePartition).nextToken(null).build()));
+
+ ArgumentCaptor<BatchUpdatePartitionRequest> captor =
ArgumentCaptor.forClass(BatchUpdatePartitionRequest.class);
+ when(mockAwsGlue.batchUpdatePartition(captor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(BatchUpdatePartitionResponse.builder().build()));
+
+ HoodieSchema schema = GlueTestUtil.getSimpleSchema();
+ SchemaDifference schemaDiff = SchemaDifference.newBuilder(schema, new
HashMap<>())
+ .updateTableColumn("name", "string")
+ .build();
+
+ awsGlueSyncClient.updateTableSchema(tableName, schema, schemaDiff);
+
+ List<BatchUpdatePartitionRequestEntry> entries =
captor.getValue().entries();
+ assertEquals(1, entries.size());
+ assertEquals(partitionLocation,
entries.get(0).partitionInput().storageDescriptor().location());
+ assertEquals(Collections.singletonList("2024-01-15"),
entries.get(0).partitionValueList());
+ assertEquals(updatedTable.storageDescriptor().columns(),
+ entries.get(0).partitionInput().storageDescriptor().columns());
+
+ InOrder inOrder = inOrder(mockAwsGlue);
+ inOrder.verify(mockAwsGlue).updateTable(any(UpdateTableRequest.class));
+
inOrder.verify(mockAwsGlue).batchUpdatePartition(any(BatchUpdatePartitionRequest.class));
+ }
}
