voonhous commented on code in PR #19876:
URL: https://github.com/apache/hudi/pull/19876#discussion_r3975337543
##########
hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java:
##########
@@ -1010,4 +1034,631 @@ void testUpdateTableSchema_issuesOneUpdateTable() {
verify(mockAwsGlue, times(1)).updateTable(any(UpdateTableRequest.class));
verify(mockAwsGlue,
never()).batchUpdatePartition(any(BatchUpdatePartitionRequest.class));
}
+
+ @Test
+ void testGetPartitionsFromList_returnsPartitionsKnownToGlue() {
+ String tableName = "tbl";
+ software.amazon.awssdk.services.glue.model.Partition gluePartition =
+ software.amazon.awssdk.services.glue.model.Partition.builder()
+ .values("2024-01-15")
+
.storageDescriptor(StorageDescriptor.builder().location("s3://base/2024/01/15").build())
+ .build();
+ ArgumentCaptor<BatchGetPartitionRequest> captor =
ArgumentCaptor.forClass(BatchGetPartitionRequest.class);
+ when(mockAwsGlue.batchGetPartition(captor.capture()))
+ .thenReturn(CompletableFuture.completedFuture(
+
BatchGetPartitionResponse.builder().partitions(gluePartition).build()));
+
+ List<Partition> result =
awsGlueSyncClient.getPartitionsFromList(tableName, Arrays.asList("2024/01/15",
"2024/01/16"));
+
+ assertEquals(1, result.size(), "only the partition Glue knows about is
returned");
+ assertEquals(Collections.singletonList("2024-01-15"),
result.get(0).getValues());
+ assertEquals("s3://base/2024/01/15", result.get(0).getStorageLocation());
+
+ BatchGetPartitionRequest sent = captor.getValue();
+ assertEquals(GlueTestUtil.DB_NAME, sent.databaseName());
+ assertEquals(tableName, sent.tableName());
+ assertEquals(Arrays.asList(Collections.singletonList("2024-01-15"),
Collections.singletonList("2024-01-16")),
+
sent.partitionsToGet().stream().map(PartitionValueList::values).collect(Collectors.toList()),
+ "the requested partitions are the extracted partition values, not the
storage paths");
+ }
+
+ @Test
+ void testGetPartitionsFromList_emptyListDoesNotCallGlue() {
+ assertTrue(awsGlueSyncClient.getPartitionsFromList("tbl",
Collections.emptyList()).isEmpty());
+ verify(mockAwsGlue,
never()).batchGetPartition(any(BatchGetPartitionRequest.class));
+ }
+
+ @Test
+ void testGetMetastoreSchema_mergesColumnsAndPartitionKeys() {
+ String tableName = "tbl";
+ List<Column> columns = Arrays.asList(GlueTestUtil.getColumn("name",
"string", null),
+ GlueTestUtil.getColumn("age", "int", null));
+ List<Column> partitionKeys =
Collections.singletonList(GlueTestUtil.getColumn("datestr", "string", null));
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(getTableWithDefaultProps(tableName, columns,
partitionKeys));
+
+ Map<String, String> schema =
awsGlueSyncClient.getMetastoreSchema(tableName);
+
+ assertEquals(3, schema.size());
+ assertEquals("STRING", schema.get("name"), "column types are upper cased");
+ assertEquals("INT", schema.get("age"));
+ assertEquals("STRING", schema.get("datestr"), "partition keys are merged
into the schema");
+ }
+
+ @Test
+ void testGetMetastoreSchema_wrapsGlueFailure() {
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.getMetastoreSchema("tbl"));
+ assertTrue(ex.getMessage().contains("Fail to get schema for table"));
+ }
+
+ @Test
+ void testGetLastCommitTimeSynced_readsTableParameters() {
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(HOODIE_LAST_COMMIT_TIME_SYNC, "100");
+ parameters.put(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC, "110");
+ Table withSyncTimes = tableWithParameters("synced", parameters);
+ Table withoutSyncTimes = tableWithParameters("unsynced", new HashMap<>());
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(withSyncTimes).build()))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(withoutSyncTimes).build()));
+
+ assertEquals("100",
awsGlueSyncClient.getLastCommitTimeSynced("synced").get());
+ assertEquals("110",
awsGlueSyncClient.getLastCommitCompletionTimeSynced("synced").get());
+ // the table is cached per name, so the second table name triggers the
second stubbed response
+
assertFalse(awsGlueSyncClient.getLastCommitTimeSynced("unsynced").isPresent());
+
assertFalse(awsGlueSyncClient.getLastCommitCompletionTimeSynced("unsynced").isPresent());
+ verify(mockAwsGlue, times(2)).getTable(any(GetTableRequest.class));
+ }
+
+ @Test
+ void testGetStorageFieldSchemas_readsFieldsAndDocsFromStorage() {
+ Map<String, FieldSchema> byName =
awsGlueSyncClient.getStorageFieldSchemas().stream()
+ .collect(Collectors.toMap(FieldSchema::getName, f -> f));
+
+ assertEquals("int", byName.get("id").getType());
+ assertEquals(GlueTestUtil.ID_FIELD_DOC,
byName.get("id").getComment().get());
+ assertEquals("string", byName.get("name").getType());
+ assertEquals(GlueTestUtil.NAME_FIELD_DOC,
byName.get("name").getComment().get());
+ assertTrue(byName.containsKey("_hoodie_commit_time"), "metadata fields are
part of the storage schema");
+ }
+
+ @Test
+ void testManagePartitionIndexes_disabledDeactivatesFlagAndDropsIndexes()
throws Exception {
+ String tableName = "tbl";
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(GLUE_PARTITION_INDEX_ENABLE, "true");
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
parameters)).build()));
+ ArgumentCaptor<UpdateTableRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(updateCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(partitionIndexDescriptor("idx_one",
"datestr"))
+ .build()));
+ ArgumentCaptor<DeletePartitionIndexRequest> deleteCaptor =
ArgumentCaptor.forClass(DeletePartitionIndexRequest.class);
+ when(mockAwsGlue.deletePartitionIndex(deleteCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(DeletePartitionIndexResponse.builder().build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ assertEquals("false",
updateCaptor.getValue().tableInput().parameters().get(GLUE_PARTITION_INDEX_ENABLE),
+ "partition index usage is deactivated when the feature is off");
+ assertEquals(Collections.singletonList("idx_one"),
deleteCaptor.getAllValues().stream()
+
.map(DeletePartitionIndexRequest::indexName).collect(Collectors.toList()));
+ verify(mockAwsGlue,
never()).createPartitionIndex(any(CreatePartitionIndexRequest.class));
+ }
+
+ @Test
+ void
testManagePartitionIndexes_enabledDropsStaleIndexesAndCreatesMissingOnes()
throws Exception {
+ String tableName = "tbl";
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS_ENABLE.key(),
"true");
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS.key(),
"datestr;hour,region");
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), GlueTestUtil.getMetaClient());
+
+ // the table has no partition_filtering.enabled parameter, so indexing has
to be activated first
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ ArgumentCaptor<UpdateTableRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(updateCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
+ PartitionIndexDescriptor keptIndex = partitionIndexDescriptor("kept_idx",
"datestr", "hour");
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(keptIndex,
partitionIndexDescriptor("stale_idx", "old_col"))
+ .build()))
+ // after a drop the index list is re-read
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(keptIndex)
+ .build()));
+ ArgumentCaptor<DeletePartitionIndexRequest> deleteCaptor =
ArgumentCaptor.forClass(DeletePartitionIndexRequest.class);
+ when(mockAwsGlue.deletePartitionIndex(deleteCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(DeletePartitionIndexResponse.builder().build()));
+ ArgumentCaptor<CreatePartitionIndexRequest> createCaptor =
ArgumentCaptor.forClass(CreatePartitionIndexRequest.class);
+ when(mockAwsGlue.createPartitionIndex(createCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(CreatePartitionIndexResponse.builder().build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ assertEquals("true",
updateCaptor.getValue().tableInput().parameters().get(GLUE_PARTITION_INDEX_ENABLE));
+ assertEquals(Collections.singletonList("stale_idx"),
deleteCaptor.getAllValues().stream()
+
.map(DeletePartitionIndexRequest::indexName).collect(Collectors.toList()),
+ "only the index that is no longer configured is dropped");
+ assertEquals(1, createCaptor.getAllValues().size(), "the already existing
index is not recreated");
+ PartitionIndex created = createCaptor.getValue().partitionIndex();
+ assertEquals(Collections.singletonList("region"), created.keys());
+ assertEquals("hudi_managed_[region]", created.indexName());
+ verify(mockAwsGlue,
times(2)).getPartitionIndexes(any(GetPartitionIndexesRequest.class));
+ }
+
+ @Test
+ void testParsePartitionsIndexConfig_keepsOnlyTheFirstThreeIndexes() {
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS.key(),
"a;b,c,d,e");
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), GlueTestUtil.getMetaClient());
+
+ assertEquals(Arrays.asList(Arrays.asList("a", "b"),
Collections.singletonList("c"), Collections.singletonList("d")),
+ awsGlueSyncClient.parsePartitionsIndexConfig(), "glue supports at most
three partition indexes");
+ }
+
+ @Test
+ void testUpdateLastCommitTimeSynced_writesTimelineInstantToTableParameters()
{
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ ArgumentCaptor<UpdateTableRequest> captor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(captor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder().build()));
+
+ awsGlueSyncClient.updateLastCommitTimeSynced(tableName);
+
+ Map<String, String> parameters =
captor.getValue().tableInput().parameters();
+ assertEquals(GlueTestUtil.INSTANT_TIME,
parameters.get(HOODIE_LAST_COMMIT_TIME_SYNC),
+ "the last instant of the active timeline is synced");
+ assertEquals(GlueTestUtil.COMPLETION_TIME,
parameters.get(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC),
+ "the completion time of that instant is synced alongside it");
+ assertTrue(captor.getValue().skipArchive(), "table archiving is skipped by
default");
+ }
+
+ /**
+ * An indexation already in flight surfaces as an {@link
ExecutionException}, anything else lands in the
+ * catch-all. Neither may fail the commit-time sync.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void
testUpdateLastCommitTimeSynced_partitionIndexFailureDoesNotFailTheSync(boolean
asExecutionFailure) throws Exception {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ when(mockAwsGlue.updateTable(any(UpdateTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+ if (asExecutionFailure) {
+ CompletableFuture<GetPartitionIndexesResponse> failed =
mock(CompletableFuture.class);
+ when(failed.get()).thenThrow(new ExecutionException(new
RuntimeException("indexing in progress")));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class))).thenReturn(failed);
+ } else {
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class))).thenThrow(new
RuntimeException("boom"));
+ }
+
+ awsGlueSyncClient.updateLastCommitTimeSynced(tableName);
+
+ verify(mockAwsGlue, times(1)).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateLastCommitTimeSynced_wrapsGlueFailure() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ when(mockAwsGlue.updateTable(any(UpdateTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.updateLastCommitTimeSynced(tableName));
+ assertTrue(ex.getMessage().contains("Fail to update last sync commit
time"));
+ }
+
+ @Test
+ void testUpdateSerdeProperties_emptyPropertiesSkipUpdate() {
+ assertFalse(awsGlueSyncClient.updateSerdeProperties("tbl",
Collections.emptyMap(), false));
+ verify(mockAwsGlue, never()).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateSerdeProperties_unchangedPropertiesSkipUpdate() {
+ String tableName = "tbl";
+ Map<String, String> serdeProperties = new HashMap<>();
+ serdeProperties.put("serialization.format", "1");
+ serdeProperties.put("path", "s3://base");
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+
GetTableResponse.builder().table(tableWithSerdeProperties(tableName,
serdeProperties)).build()));
+
+ assertFalse(awsGlueSyncClient.updateSerdeProperties(tableName, new
HashMap<>(serdeProperties), false));
+ verify(mockAwsGlue, never()).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateSerdeProperties_changedPropertiesRewriteSerdeInfo() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+
GetTableResponse.builder().table(tableWithSerdeProperties(tableName,
+ serdePropertiesOf("serialization.format", "1", "location",
"s3://old"))).build()));
+ ArgumentCaptor<UpdateTableRequest> captor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(captor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
+ Map<String, String> serdeProperties = new HashMap<>();
+ serdeProperties.put("path", "s3://new");
+ assertTrue(awsGlueSyncClient.updateSerdeProperties(tableName,
serdeProperties, false));
Review Comment:
It is not read anywhere in the Glue client; only `HoodieHiveSyncClient` uses
it to pick the input format. The changed-properties test is now parameterized
over both values and asserts the same request, with the input and output format
carried over from the catalog unchanged. Whether Glue should honour the flag
for `_rt` tables is a separate question, so production is untouched here. Done
in bb7fccea3715.
##########
hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAWSGlueSyncClient.java:
##########
@@ -1010,4 +1034,631 @@ void testUpdateTableSchema_issuesOneUpdateTable() {
verify(mockAwsGlue, times(1)).updateTable(any(UpdateTableRequest.class));
verify(mockAwsGlue,
never()).batchUpdatePartition(any(BatchUpdatePartitionRequest.class));
}
+
+ @Test
+ void testGetPartitionsFromList_returnsPartitionsKnownToGlue() {
+ String tableName = "tbl";
+ software.amazon.awssdk.services.glue.model.Partition gluePartition =
+ software.amazon.awssdk.services.glue.model.Partition.builder()
+ .values("2024-01-15")
+
.storageDescriptor(StorageDescriptor.builder().location("s3://base/2024/01/15").build())
+ .build();
+ ArgumentCaptor<BatchGetPartitionRequest> captor =
ArgumentCaptor.forClass(BatchGetPartitionRequest.class);
+ when(mockAwsGlue.batchGetPartition(captor.capture()))
+ .thenReturn(CompletableFuture.completedFuture(
+
BatchGetPartitionResponse.builder().partitions(gluePartition).build()));
+
+ List<Partition> result =
awsGlueSyncClient.getPartitionsFromList(tableName, Arrays.asList("2024/01/15",
"2024/01/16"));
+
+ assertEquals(1, result.size(), "only the partition Glue knows about is
returned");
+ assertEquals(Collections.singletonList("2024-01-15"),
result.get(0).getValues());
+ assertEquals("s3://base/2024/01/15", result.get(0).getStorageLocation());
+
+ BatchGetPartitionRequest sent = captor.getValue();
+ assertEquals(GlueTestUtil.DB_NAME, sent.databaseName());
+ assertEquals(tableName, sent.tableName());
+ assertEquals(Arrays.asList(Collections.singletonList("2024-01-15"),
Collections.singletonList("2024-01-16")),
+
sent.partitionsToGet().stream().map(PartitionValueList::values).collect(Collectors.toList()),
+ "the requested partitions are the extracted partition values, not the
storage paths");
+ }
+
+ @Test
+ void testGetPartitionsFromList_emptyListDoesNotCallGlue() {
+ assertTrue(awsGlueSyncClient.getPartitionsFromList("tbl",
Collections.emptyList()).isEmpty());
+ verify(mockAwsGlue,
never()).batchGetPartition(any(BatchGetPartitionRequest.class));
+ }
+
+ @Test
+ void testGetMetastoreSchema_mergesColumnsAndPartitionKeys() {
+ String tableName = "tbl";
+ List<Column> columns = Arrays.asList(GlueTestUtil.getColumn("name",
"string", null),
+ GlueTestUtil.getColumn("age", "int", null));
+ List<Column> partitionKeys =
Collections.singletonList(GlueTestUtil.getColumn("datestr", "string", null));
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(getTableWithDefaultProps(tableName, columns,
partitionKeys));
+
+ Map<String, String> schema =
awsGlueSyncClient.getMetastoreSchema(tableName);
+
+ assertEquals(3, schema.size());
+ assertEquals("STRING", schema.get("name"), "column types are upper cased");
+ assertEquals("INT", schema.get("age"));
+ assertEquals("STRING", schema.get("datestr"), "partition keys are merged
into the schema");
+ }
+
+ @Test
+ void testGetMetastoreSchema_wrapsGlueFailure() {
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.getMetastoreSchema("tbl"));
+ assertTrue(ex.getMessage().contains("Fail to get schema for table"));
+ }
+
+ @Test
+ void testGetLastCommitTimeSynced_readsTableParameters() {
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(HOODIE_LAST_COMMIT_TIME_SYNC, "100");
+ parameters.put(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC, "110");
+ Table withSyncTimes = tableWithParameters("synced", parameters);
+ Table withoutSyncTimes = tableWithParameters("unsynced", new HashMap<>());
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(withSyncTimes).build()))
+
.thenReturn(CompletableFuture.completedFuture(GetTableResponse.builder().table(withoutSyncTimes).build()));
+
+ assertEquals("100",
awsGlueSyncClient.getLastCommitTimeSynced("synced").get());
+ assertEquals("110",
awsGlueSyncClient.getLastCommitCompletionTimeSynced("synced").get());
+ // the table is cached per name, so the second table name triggers the
second stubbed response
+
assertFalse(awsGlueSyncClient.getLastCommitTimeSynced("unsynced").isPresent());
+
assertFalse(awsGlueSyncClient.getLastCommitCompletionTimeSynced("unsynced").isPresent());
+ verify(mockAwsGlue, times(2)).getTable(any(GetTableRequest.class));
+ }
+
+ @Test
+ void testGetStorageFieldSchemas_readsFieldsAndDocsFromStorage() {
+ Map<String, FieldSchema> byName =
awsGlueSyncClient.getStorageFieldSchemas().stream()
+ .collect(Collectors.toMap(FieldSchema::getName, f -> f));
+
+ assertEquals("int", byName.get("id").getType());
+ assertEquals(GlueTestUtil.ID_FIELD_DOC,
byName.get("id").getComment().get());
+ assertEquals("string", byName.get("name").getType());
+ assertEquals(GlueTestUtil.NAME_FIELD_DOC,
byName.get("name").getComment().get());
+ assertTrue(byName.containsKey("_hoodie_commit_time"), "metadata fields are
part of the storage schema");
+ }
+
+ @Test
+ void testManagePartitionIndexes_disabledDeactivatesFlagAndDropsIndexes()
throws Exception {
+ String tableName = "tbl";
+ Map<String, String> parameters = new HashMap<>();
+ parameters.put(GLUE_PARTITION_INDEX_ENABLE, "true");
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
parameters)).build()));
+ ArgumentCaptor<UpdateTableRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(updateCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(partitionIndexDescriptor("idx_one",
"datestr"))
+ .build()));
+ ArgumentCaptor<DeletePartitionIndexRequest> deleteCaptor =
ArgumentCaptor.forClass(DeletePartitionIndexRequest.class);
+ when(mockAwsGlue.deletePartitionIndex(deleteCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(DeletePartitionIndexResponse.builder().build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ assertEquals("false",
updateCaptor.getValue().tableInput().parameters().get(GLUE_PARTITION_INDEX_ENABLE),
+ "partition index usage is deactivated when the feature is off");
+ assertEquals(Collections.singletonList("idx_one"),
deleteCaptor.getAllValues().stream()
+
.map(DeletePartitionIndexRequest::indexName).collect(Collectors.toList()));
+ verify(mockAwsGlue,
never()).createPartitionIndex(any(CreatePartitionIndexRequest.class));
+ }
+
+ @Test
+ void
testManagePartitionIndexes_enabledDropsStaleIndexesAndCreatesMissingOnes()
throws Exception {
+ String tableName = "tbl";
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS_ENABLE.key(),
"true");
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS.key(),
"datestr;hour,region");
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), GlueTestUtil.getMetaClient());
+
+ // the table has no partition_filtering.enabled parameter, so indexing has
to be activated first
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ ArgumentCaptor<UpdateTableRequest> updateCaptor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(updateCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
+ PartitionIndexDescriptor keptIndex = partitionIndexDescriptor("kept_idx",
"datestr", "hour");
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(keptIndex,
partitionIndexDescriptor("stale_idx", "old_col"))
+ .build()))
+ // after a drop the index list is re-read
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder()
+ .partitionIndexDescriptorList(keptIndex)
+ .build()));
+ ArgumentCaptor<DeletePartitionIndexRequest> deleteCaptor =
ArgumentCaptor.forClass(DeletePartitionIndexRequest.class);
+ when(mockAwsGlue.deletePartitionIndex(deleteCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(DeletePartitionIndexResponse.builder().build()));
+ ArgumentCaptor<CreatePartitionIndexRequest> createCaptor =
ArgumentCaptor.forClass(CreatePartitionIndexRequest.class);
+ when(mockAwsGlue.createPartitionIndex(createCaptor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(CreatePartitionIndexResponse.builder().build()));
+
+ awsGlueSyncClient.managePartitionIndexes(tableName);
+
+ assertEquals("true",
updateCaptor.getValue().tableInput().parameters().get(GLUE_PARTITION_INDEX_ENABLE));
+ assertEquals(Collections.singletonList("stale_idx"),
deleteCaptor.getAllValues().stream()
+
.map(DeletePartitionIndexRequest::indexName).collect(Collectors.toList()),
+ "only the index that is no longer configured is dropped");
+ assertEquals(1, createCaptor.getAllValues().size(), "the already existing
index is not recreated");
+ PartitionIndex created = createCaptor.getValue().partitionIndex();
+ assertEquals(Collections.singletonList("region"), created.keys());
+ assertEquals("hudi_managed_[region]", created.indexName());
+ verify(mockAwsGlue,
times(2)).getPartitionIndexes(any(GetPartitionIndexesRequest.class));
+ }
+
+ @Test
+ void testParsePartitionsIndexConfig_keepsOnlyTheFirstThreeIndexes() {
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+
props.setProperty(GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS.key(),
"a;b,c,d,e");
+ awsGlueSyncClient = new AWSGlueCatalogSyncClient(mockAwsGlue, mockSts, new
HiveSyncConfig(props), GlueTestUtil.getMetaClient());
+
+ assertEquals(Arrays.asList(Arrays.asList("a", "b"),
Collections.singletonList("c"), Collections.singletonList("d")),
+ awsGlueSyncClient.parsePartitionsIndexConfig(), "glue supports at most
three partition indexes");
+ }
+
+ @Test
+ void testUpdateLastCommitTimeSynced_writesTimelineInstantToTableParameters()
{
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ ArgumentCaptor<UpdateTableRequest> captor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(captor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(GetPartitionIndexesResponse.builder().build()));
+
+ awsGlueSyncClient.updateLastCommitTimeSynced(tableName);
+
+ Map<String, String> parameters =
captor.getValue().tableInput().parameters();
+ assertEquals(GlueTestUtil.INSTANT_TIME,
parameters.get(HOODIE_LAST_COMMIT_TIME_SYNC),
+ "the last instant of the active timeline is synced");
+ assertEquals(GlueTestUtil.COMPLETION_TIME,
parameters.get(HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC),
+ "the completion time of that instant is synced alongside it");
+ assertTrue(captor.getValue().skipArchive(), "table archiving is skipped by
default");
+ }
+
+ /**
+ * An indexation already in flight surfaces as an {@link
ExecutionException}, anything else lands in the
+ * catch-all. Neither may fail the commit-time sync.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void
testUpdateLastCommitTimeSynced_partitionIndexFailureDoesNotFailTheSync(boolean
asExecutionFailure) throws Exception {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ when(mockAwsGlue.updateTable(any(UpdateTableRequest.class)))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+ if (asExecutionFailure) {
+ CompletableFuture<GetPartitionIndexesResponse> failed =
mock(CompletableFuture.class);
+ when(failed.get()).thenThrow(new ExecutionException(new
RuntimeException("indexing in progress")));
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class))).thenReturn(failed);
+ } else {
+
when(mockAwsGlue.getPartitionIndexes(any(GetPartitionIndexesRequest.class))).thenThrow(new
RuntimeException("boom"));
+ }
+
+ awsGlueSyncClient.updateLastCommitTimeSynced(tableName);
+
+ verify(mockAwsGlue, times(1)).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateLastCommitTimeSynced_wrapsGlueFailure() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+ GetTableResponse.builder().table(tableWithParameters(tableName,
new HashMap<>())).build()));
+ when(mockAwsGlue.updateTable(any(UpdateTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.updateLastCommitTimeSynced(tableName));
+ assertTrue(ex.getMessage().contains("Fail to update last sync commit
time"));
+ }
+
+ @Test
+ void testUpdateSerdeProperties_emptyPropertiesSkipUpdate() {
+ assertFalse(awsGlueSyncClient.updateSerdeProperties("tbl",
Collections.emptyMap(), false));
+ verify(mockAwsGlue, never()).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateSerdeProperties_unchangedPropertiesSkipUpdate() {
+ String tableName = "tbl";
+ Map<String, String> serdeProperties = new HashMap<>();
+ serdeProperties.put("serialization.format", "1");
+ serdeProperties.put("path", "s3://base");
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+
GetTableResponse.builder().table(tableWithSerdeProperties(tableName,
serdeProperties)).build()));
+
+ assertFalse(awsGlueSyncClient.updateSerdeProperties(tableName, new
HashMap<>(serdeProperties), false));
+ verify(mockAwsGlue, never()).updateTable(any(UpdateTableRequest.class));
+ }
+
+ @Test
+ void testUpdateSerdeProperties_changedPropertiesRewriteSerdeInfo() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(CompletableFuture.completedFuture(
+
GetTableResponse.builder().table(tableWithSerdeProperties(tableName,
+ serdePropertiesOf("serialization.format", "1", "location",
"s3://old"))).build()));
+ ArgumentCaptor<UpdateTableRequest> captor =
ArgumentCaptor.forClass(UpdateTableRequest.class);
+ when(mockAwsGlue.updateTable(captor.capture()))
+
.thenReturn(CompletableFuture.completedFuture(UpdateTableResponse.builder().build()));
+
+ Map<String, String> serdeProperties = new HashMap<>();
+ serdeProperties.put("path", "s3://new");
+ assertTrue(awsGlueSyncClient.updateSerdeProperties(tableName,
serdeProperties, false));
+
+ SerDeInfo sent =
captor.getValue().tableInput().storageDescriptor().serdeInfo();
+
assertEquals("org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe",
sent.serializationLibrary(),
+ "the serde class is derived from the base file format");
+ assertEquals("s3://new", sent.parameters().get("path"));
+ assertEquals("1", sent.parameters().get("serialization.format"), "the
serialization format is defaulted in");
+ }
+
+ @Test
+ void testUpdateSerdeProperties_wrapsGlueFailure() {
+
when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(EntityNotFoundException.class);
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
+ () -> awsGlueSyncClient.updateSerdeProperties("tbl", new
HashMap<>(Collections.singletonMap("path", "s3://new")), false));
+ assertTrue(ex.getMessage().contains("Failed to update table serde info for
table"));
+ }
+
+ @Test
+ void testCreateTable_existingTableIsNotRecreated() {
+ String tableName = "tbl";
+ when(mockAwsGlue.getTable(any(GetTableRequest.class)))
+ .thenReturn(getTableWithDefaultProps(tableName,
Collections.emptyList(), Collections.emptyList()));
+
+ awsGlueSyncClient.createTable(tableName, GlueTestUtil.getSimpleSchema(),
"inputFormat", "outputFormat",
+ "serde", new HashMap<>(), new HashMap<>());
+
+ verify(mockAwsGlue, never()).createTable(any(CreateTableRequest.class));
+ }
+
+ @Test
+ void testTableExists_wrapsNonEntityNotFoundExecutionFailure() throws
Exception {
+ CompletableFuture<GetTableResponse> failed = mock(CompletableFuture.class);
+ when(failed.get()).thenThrow(new ExecutionException(new
RuntimeException("boom")));
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenReturn(failed);
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
() -> awsGlueSyncClient.tableExists("tbl"));
+ assertTrue(ex.getMessage().contains("Fail to get table"));
+ }
+
+ @Test
+ void testTableExists_wrapsClientFailure() {
+ when(mockAwsGlue.getTable(any(GetTableRequest.class))).thenThrow(new
RuntimeException("boom"));
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
() -> awsGlueSyncClient.tableExists("tbl"));
+ assertTrue(ex.getMessage().contains("Fail to get table"));
+ }
+
+ @Test
+ void testDatabaseExists_wrapsClientFailure() {
+ when(mockAwsGlue.getDatabase(any(GetDatabaseRequest.class))).thenThrow(new
RuntimeException("boom"));
+
+ HoodieGlueSyncException ex = assertThrows(HoodieGlueSyncException.class,
() -> awsGlueSyncClient.databaseExists("db"));
+ assertTrue(ex.getMessage().contains("Fail to check if database exists"));
+ }
+
+ @Test
+ void testDropTable_interruptionRestoresTheInterruptFlag() throws Exception {
+ CompletableFuture<DeleteTableResponse> failed =
mock(CompletableFuture.class);
+ when(failed.get()).thenThrow(new InterruptedException("interrupted"));
+
when(mockAwsGlue.deleteTable(any(DeleteTableRequest.class))).thenReturn(failed);
+
+ assertThrows(HoodieGlueSyncException.class, () ->
awsGlueSyncClient.dropTable("tbl"));
+ assertTrue(Thread.interrupted(), "the interrupt flag is restored for
handlers up the stack");
+ }
+
+ @Test
+ void testBuildAsyncClient_appliesTheConfiguredEndpointAndRegion() {
+ TypedProperties props = GlueTestUtil.getHiveSyncConfig().getProps();
+ props.setProperty(HoodieAWSConfig.AWS_GLUE_ENDPOINT.key(),
"https://glue.eu-west-1.amazonaws.com";);
+ props.setProperty(HoodieAWSConfig.AWS_GLUE_REGION.key(), "eu-west-1");
+
+ try (MockedStatic<GlueAsyncClient> glueStatic =
mockStatic(GlueAsyncClient.class);
+ MockedStatic<StsClient> stsStatic = mockStatic(StsClient.class)) {
+ GlueAsyncClientBuilder builder = mock(GlueAsyncClientBuilder.class);
+ glueStatic.when(GlueAsyncClient::builder).thenReturn(builder);
+ when(builder.credentialsProvider(any())).thenReturn(builder);
+ when(builder.endpointOverride(any(URI.class))).thenReturn(builder);
+ when(builder.region(any(Region.class))).thenReturn(builder);
+ when(builder.build()).thenReturn(mockAwsGlue);
+ stsStatic.when(StsClient::create).thenReturn(mockSts);
+
+ new AWSGlueCatalogSyncClient(new HiveSyncConfig(props),
GlueTestUtil.getMetaClient());
+
+
verify(builder).endpointOverride(URI.create("https://glue.eu-west-1.amazonaws.com";));
Review Comment:
Added `testBuildAsyncClient_withoutAnEndpointOrRegionKeepsTheSdkDefaults`:
neither property set, `credentialsProvider` and `build` called,
`endpointOverride` and `region` never. The builder stubbing moved into a shared
`mockGlueClientBuilder` helper used by all three tests. Done in bb7fccea3715.
--
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]
