You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2024/05/14 22:08:42 UTC
(hudi) 06/28: [HUDI-7466] Add parallel listing of existing partitions in Glue Catalog sync (#10460)
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 41ba99d5e5ba77ded18f0a398f54de40eab8cbca
Author: Vitali Makarevich <vi...@gmail.com>
AuthorDate: Sat Mar 16 02:06:23 2024 +0100
[HUDI-7466] Add parallel listing of existing partitions in Glue Catalog sync (#10460)
* Add parallel listing of existing partitions
* Improve with new approach
* Fix checkstyle
* Fix listing for empty list of commits
* Fix logic for HiveSyncTool
* Fix lint errors
* Fix IT
* Use custom thread names
* Address review comments
---------
Co-authored-by: vmakarevich <vi...@instructure.com>
---
.../hudi/aws/sync/AWSGlueCatalogSyncClient.java | 301 ++++++++++++++-------
.../hudi/config/GlueCatalogSyncClientConfig.java | 24 ++
.../hudi/aws/sync/ITTestGluePartitionPushdown.java | 31 +--
.../java/org/apache/hudi/hive/HiveSyncTool.java | 14 +-
.../org/apache/hudi/hive/HoodieHiveSyncClient.java | 14 +-
.../hudi/sync/common/HoodieMetaSyncOperations.java | 6 +-
6 files changed, 245 insertions(+), 145 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 15847129d8a..5f2fc3cefdc 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
@@ -22,6 +22,8 @@ import org.apache.hudi.aws.sync.util.GluePartitionFilterGenerator;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.table.TableSchemaResolver;
import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.CustomizedThreadFactory;
+import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.GlueCatalogSyncClientConfig;
import org.apache.hudi.hive.HiveSyncConfig;
@@ -37,6 +39,8 @@ import software.amazon.awssdk.services.glue.model.BatchCreatePartitionRequest;
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.BatchGetPartitionRequest;
+import software.amazon.awssdk.services.glue.model.BatchGetPartitionResponse;
import software.amazon.awssdk.services.glue.model.BatchUpdatePartitionRequest;
import software.amazon.awssdk.services.glue.model.BatchUpdatePartitionRequestEntry;
import software.amazon.awssdk.services.glue.model.BatchUpdatePartitionResponse;
@@ -59,6 +63,7 @@ import software.amazon.awssdk.services.glue.model.PartitionIndex;
import software.amazon.awssdk.services.glue.model.PartitionIndexDescriptor;
import software.amazon.awssdk.services.glue.model.PartitionInput;
import software.amazon.awssdk.services.glue.model.PartitionValueList;
+import software.amazon.awssdk.services.glue.model.Segment;
import software.amazon.awssdk.services.glue.model.SerDeInfo;
import software.amazon.awssdk.services.glue.model.StorageDescriptor;
import software.amazon.awssdk.services.glue.model.Table;
@@ -81,14 +86,21 @@ import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.function.Consumer;
import java.util.stream.Collectors;
import static org.apache.hudi.aws.utils.S3Utils.s3aToS3;
import static org.apache.hudi.common.util.MapUtils.containsAll;
import static org.apache.hudi.common.util.MapUtils.isNullOrEmpty;
-import static org.apache.hudi.config.GlueCatalogSyncClientConfig.GLUE_METADATA_FILE_LISTING;
+import static org.apache.hudi.config.GlueCatalogSyncClientConfig.CHANGED_PARTITIONS_READ_PARALLELISM;
import static org.apache.hudi.config.GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS;
import static org.apache.hudi.config.GlueCatalogSyncClientConfig.META_SYNC_PARTITION_INDEX_FIELDS_ENABLE;
+import static org.apache.hudi.config.GlueCatalogSyncClientConfig.PARTITION_CHANGE_PARALLELISM;
+import static org.apache.hudi.config.GlueCatalogSyncClientConfig.GLUE_METADATA_FILE_LISTING;
+import static org.apache.hudi.config.GlueCatalogSyncClientConfig.ALL_PARTITIONS_READ_PARALLELISM;
import static org.apache.hudi.config.HoodieAWSConfig.AWS_GLUE_ENDPOINT;
import static org.apache.hudi.config.HoodieAWSConfig.AWS_GLUE_REGION;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_CREATE_MANAGED_TABLE;
@@ -108,12 +120,12 @@ import static org.apache.hudi.sync.common.util.TableUtils.tableId;
public class AWSGlueCatalogSyncClient extends HoodieSyncClient {
private static final Logger LOG = LoggerFactory.getLogger(AWSGlueCatalogSyncClient.class);
- private static final int MAX_PARTITIONS_PER_REQUEST = 100;
+ private static final int MAX_PARTITIONS_PER_CHANGE_REQUEST = 100;
+ private static final int MAX_PARTITIONS_PER_READ_REQUEST = 1000;
private static final int MAX_DELETE_PARTITIONS_PER_REQUEST = 25;
protected final GlueAsyncClient awsGlue;
private static final String GLUE_PARTITION_INDEX_ENABLE = "partition_filtering.enabled";
private static final int PARTITION_INDEX_MAX_NUMBER = 3;
- private static final int GLUE_EXPRESSION_MAX_CHARS = 2048;
/**
* athena v2/v3 table property
* see https://docs.aws.amazon.com/athena/latest/ug/querying-hudi.html
@@ -123,6 +135,9 @@ public class AWSGlueCatalogSyncClient extends HoodieSyncClient {
private final Boolean skipTableArchive;
private final String enableMetadataTable;
+ private final int allPartitionsReadParallelism;
+ private final int changedPartitionsReadParallelism;
+ private final int changeParallelism;
public AWSGlueCatalogSyncClient(HiveSyncConfig config) {
super(config);
@@ -139,105 +154,196 @@ public class AWSGlueCatalogSyncClient extends HoodieSyncClient {
this.databaseName = config.getStringOrDefault(META_SYNC_DATABASE_NAME);
this.skipTableArchive = config.getBooleanOrDefault(GlueCatalogSyncClientConfig.GLUE_SKIP_TABLE_ARCHIVE);
this.enableMetadataTable = Boolean.toString(config.getBoolean(GLUE_METADATA_FILE_LISTING)).toUpperCase();
+ this.allPartitionsReadParallelism = config.getIntOrDefault(ALL_PARTITIONS_READ_PARALLELISM);
+ this.changedPartitionsReadParallelism = config.getIntOrDefault(CHANGED_PARTITIONS_READ_PARALLELISM);
+ this.changeParallelism = config.getIntOrDefault(PARTITION_CHANGE_PARALLELISM);
+ }
+
+ private List<Partition> getPartitionsSegment(Segment segment, String tableName) {
+ try {
+ List<Partition> partitions = new ArrayList<>();
+ String nextToken = null;
+ do {
+ GetPartitionsResponse result = awsGlue.getPartitions(GetPartitionsRequest.builder()
+ .databaseName(databaseName)
+ .tableName(tableName)
+ .segment(segment)
+ .nextToken(nextToken)
+ .build()).get();
+ partitions.addAll(result.partitions().stream()
+ .map(p -> new Partition(p.values(), p.storageDescriptor().location()))
+ .collect(Collectors.toList()));
+ nextToken = result.nextToken();
+ } while (nextToken != null);
+ return partitions;
+ } catch (Exception e) {
+ throw new HoodieGlueSyncException("Failed to get all partitions for table " + tableId(databaseName, tableName), e);
+ }
}
@Override
public List<Partition> getAllPartitions(String tableName) {
+ ExecutorService executorService = Executors.newFixedThreadPool(this.allPartitionsReadParallelism, new CustomizedThreadFactory("glue-sync-all-partitions", true));
try {
- return getPartitions(GetPartitionsRequest.builder()
- .databaseName(databaseName)
- .tableName(tableName));
+ List<Segment> segments = new ArrayList<>();
+ for (int i = 0; i < allPartitionsReadParallelism; i++) {
+ segments.add(Segment.builder()
+ .segmentNumber(i)
+ .totalSegments(allPartitionsReadParallelism).build());
+ }
+ List<Future<List<Partition>>> futures = segments.stream()
+ .map(segment -> executorService.submit(() -> this.getPartitionsSegment(segment, tableName)))
+ .collect(Collectors.toList());
+
+ List<Partition> partitions = new ArrayList<>();
+ for (Future<List<Partition>> future : futures) {
+ partitions.addAll(future.get());
+ }
+
+ return partitions;
} catch (Exception e) {
throw new HoodieGlueSyncException("Failed to get all partitions for table " + tableId(databaseName, tableName), e);
+ } finally {
+ executorService.shutdownNow();
}
}
@Override
- public List<Partition> getPartitionsByFilter(String tableName, String filter) {
+ public List<Partition> getPartitionsFromList(String tableName, List<String> partitionList) {
+ if (partitionList.isEmpty()) {
+ LOG.info("No partitions to read for " + tableId(this.databaseName, tableName));
+ return Collections.emptyList();
+ }
+ HoodieTimer timer = HoodieTimer.start();
+ List<List<String>> batches = CollectionUtils.batches(partitionList, MAX_PARTITIONS_PER_READ_REQUEST);
+ ExecutorService executorService = Executors.newFixedThreadPool(
+ Math.min(this.changedPartitionsReadParallelism, batches.size()),
+ new CustomizedThreadFactory("glue-sync-get-partitions-" + tableName, true)
+ );
try {
- if (filter.length() <= GLUE_EXPRESSION_MAX_CHARS) {
- LOG.info("Pushdown filters: {}", filter);
- return getPartitions(GetPartitionsRequest.builder()
- .databaseName(databaseName)
- .tableName(tableName)
- .expression(filter));
- } else {
- LOG.warn("Falling back to listing all partition since expression filter length > {}", GLUE_EXPRESSION_MAX_CHARS);
- return getAllPartitions(tableName);
+ List<Future<List<Partition>>> futures = batches
+ .stream()
+ .map(batch -> executorService.submit(() -> this.getChangedPartitions(batch, tableName)))
+ .collect(Collectors.toList());
+
+ List<Partition> partitions = new ArrayList<>();
+ for (Future<List<Partition>> future : futures) {
+ partitions.addAll(future.get());
}
+ LOG.info(
+ "Requested {} partitions, found existing {} partitions, new {} partitions",
+ partitionList.size(),
+ partitions.size(),
+ partitionList.size() - partitions.size());
+
+ return partitions;
} catch (Exception e) {
- throw new HoodieGlueSyncException("Failed to get partitions for table " + tableId(databaseName, tableName) + " from expression: " + filter, e);
+ throw new HoodieGlueSyncException("Failed to get all partitions for table " + tableId(this.databaseName, tableName), e);
+ } finally {
+ executorService.shutdownNow();
+ LOG.info("Took {} ms to get {} partitions for table {}", timer.endTimer(), partitionList.size(), tableId(this.databaseName, tableName));
}
}
- private List<Partition> getPartitions(GetPartitionsRequest.Builder partitionRequestBuilder) throws InterruptedException, ExecutionException {
- List<Partition> partitions = new ArrayList<>();
- String nextToken = null;
- do {
- GetPartitionsResponse result = awsGlue.getPartitions(partitionRequestBuilder
- .excludeColumnSchema(true)
- .nextToken(nextToken)
- .build()).get();
- partitions.addAll(result.partitions().stream()
- .map(p -> new Partition(p.values(), p.storageDescriptor().location()))
- .collect(Collectors.toList()));
- nextToken = result.nextToken();
- } while (nextToken != null);
- return partitions;
+ private List<Partition> getChangedPartitions(List<String> changedPartitions, String tableName) throws ExecutionException, InterruptedException {
+ List<PartitionValueList> partitionValueList = changedPartitions.stream().map(str ->
+ PartitionValueList.builder().values(partitionValueExtractor.extractPartitionValuesInPath(str)).build()
+ ).collect(Collectors.toList());
+ BatchGetPartitionRequest request = BatchGetPartitionRequest.builder()
+ .databaseName(this.databaseName)
+ .tableName(tableName)
+ .partitionsToGet(partitionValueList)
+ .build();
+ BatchGetPartitionResponse callResult = awsGlue.batchGetPartition(request).get();
+ List<Partition> result = callResult
+ .partitions()
+ .stream()
+ .map(p -> new Partition(p.values(), p.storageDescriptor().location()))
+ .collect(Collectors.toList());
+
+ return result;
}
@Override
public void addPartitionsToTable(String tableName, List<String> partitionsToAdd) {
- if (partitionsToAdd.isEmpty()) {
- LOG.info("No partitions to add for " + tableId(databaseName, tableName));
- return;
- }
- LOG.info("Adding " + partitionsToAdd.size() + " partition(s) in table " + tableId(databaseName, tableName));
+ HoodieTimer timer = HoodieTimer.start();
try {
+ if (partitionsToAdd.isEmpty()) {
+ LOG.info("No partitions to add for " + tableId(this.databaseName, tableName));
+ return;
+ }
Table table = getTable(awsGlue, databaseName, tableName);
+ parallelizeChange(partitionsToAdd, this.changeParallelism, partitions -> this.addPartitionsToTableInternal(table, partitions), MAX_PARTITIONS_PER_CHANGE_REQUEST);
+ } finally {
+ LOG.info("Added {} partitions to table {} in {} ms", partitionsToAdd.size(), tableId(this.databaseName, tableName), timer.endTimer());
+ }
+ }
+
+ private <T> void parallelizeChange(List<T> items, int parallelism, Consumer<List<T>> consumer, int sliceSize) {
+ List<List<T>> batches = CollectionUtils.batches(items, sliceSize);
+ ExecutorService executorService = Executors.newFixedThreadPool(Math.min(parallelism, batches.size()), new CustomizedThreadFactory("glue-sync", true));
+ try {
+ List<Future<?>> futures = batches.stream()
+ .map(item -> executorService.submit(() -> {
+ consumer.accept(item);
+ }))
+ .collect(Collectors.toList());
+ for (Future<?> future : futures) {
+ future.get();
+ }
+ } catch (Exception e) {
+ throw new HoodieGlueSyncException("Failed to parallelize operation", e);
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
+ private void addPartitionsToTableInternal(Table table, List<String> partitionsToAdd) {
+ try {
StorageDescriptor sd = table.storageDescriptor();
- List<PartitionInput> partitionInputs = partitionsToAdd.stream().map(partition -> {
+ List<PartitionInput> partitionInputList = partitionsToAdd.stream().map(partition -> {
String fullPartitionPath = FSUtils.getPartitionPath(s3aToS3(getBasePath()), partition).toString();
List<String> partitionValues = partitionValueExtractor.extractPartitionValuesInPath(partition);
StorageDescriptor partitionSD = sd.copy(copySd -> copySd.location(fullPartitionPath));
return PartitionInput.builder().values(partitionValues).storageDescriptor(partitionSD).build();
}).collect(Collectors.toList());
- List<CompletableFuture<BatchCreatePartitionResponse>> futures = new ArrayList<>();
-
- for (List<PartitionInput> batch : CollectionUtils.batches(partitionInputs, MAX_PARTITIONS_PER_REQUEST)) {
- BatchCreatePartitionRequest request = BatchCreatePartitionRequest.builder()
- .databaseName(databaseName).tableName(tableName).partitionInputList(batch).build();
- futures.add(awsGlue.batchCreatePartition(request));
- }
-
- for (CompletableFuture<BatchCreatePartitionResponse> future : futures) {
- BatchCreatePartitionResponse response = future.get();
- if (CollectionUtils.nonEmpty(response.errors())) {
- if (response.errors().stream()
- .allMatch(
- (error) -> "AlreadyExistsException".equals(error.errorDetail().errorCode()))) {
- LOG.warn("Partitions already exist in glue: " + response.errors());
- } else {
- throw new HoodieGlueSyncException("Fail to add partitions to " + tableId(databaseName, tableName)
+ BatchCreatePartitionRequest request = BatchCreatePartitionRequest.builder()
+ .databaseName(databaseName).tableName(table.name()).partitionInputList(partitionInputList).build();
+ CompletableFuture<BatchCreatePartitionResponse> future = awsGlue.batchCreatePartition(request);
+ BatchCreatePartitionResponse response = future.get();
+ if (CollectionUtils.nonEmpty(response.errors())) {
+ if (response.errors().stream()
+ .allMatch(
+ (error) -> "AlreadyExistsException".equals(error.errorDetail().errorCode()))) {
+ LOG.warn("Partitions already exist in glue: " + response.errors());
+ } else {
+ throw new HoodieGlueSyncException("Fail to add partitions to " + tableId(databaseName, table.name())
+ " with error(s): " + response.errors());
- }
}
}
} catch (Exception e) {
- throw new HoodieGlueSyncException("Fail to add partitions to " + tableId(databaseName, tableName), e);
+ throw new HoodieGlueSyncException("Fail to add partitions to " + tableId(databaseName, table.name()), e);
}
}
@Override
public void updatePartitionsToTable(String tableName, List<String> changedPartitions) {
- if (changedPartitions.isEmpty()) {
- LOG.info("No partitions to change for " + tableName);
- return;
- }
- LOG.info("Updating " + changedPartitions.size() + "partition(s) in table " + tableId(databaseName, tableName));
+ HoodieTimer timer = HoodieTimer.start();
try {
+ if (changedPartitions.isEmpty()) {
+ LOG.info("No partitions to update for " + tableId(this.databaseName, tableName));
+ return;
+ }
Table table = getTable(awsGlue, databaseName, tableName);
+ parallelizeChange(changedPartitions, this.changeParallelism, partitions -> this.updatePartitionsToTableInternal(table, 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) {
+ try {
StorageDescriptor sd = table.storageDescriptor();
List<BatchUpdatePartitionRequestEntry> updatePartitionEntries = changedPartitions.stream().map(partition -> {
String fullPartitionPath = FSUtils.getPartitionPath(s3aToS3(getBasePath()), partition).toString();
@@ -247,57 +353,52 @@ public class AWSGlueCatalogSyncClient extends HoodieSyncClient {
return BatchUpdatePartitionRequestEntry.builder().partitionInput(partitionInput).partitionValueList(partitionValues).build();
}).collect(Collectors.toList());
- List<CompletableFuture<BatchUpdatePartitionResponse>> futures = new ArrayList<>();
- for (List<BatchUpdatePartitionRequestEntry> batch : CollectionUtils.batches(updatePartitionEntries, MAX_PARTITIONS_PER_REQUEST)) {
- BatchUpdatePartitionRequest request = BatchUpdatePartitionRequest.builder()
- .databaseName(databaseName).tableName(tableName).entries(batch).build();
- futures.add(awsGlue.batchUpdatePartition(request));
- }
+ BatchUpdatePartitionRequest request = BatchUpdatePartitionRequest.builder()
+ .databaseName(databaseName).tableName(table.name()).entries(updatePartitionEntries).build();
+ CompletableFuture<BatchUpdatePartitionResponse> future = awsGlue.batchUpdatePartition(request);
- for (CompletableFuture<BatchUpdatePartitionResponse> future : futures) {
- BatchUpdatePartitionResponse response = future.get();
- if (CollectionUtils.nonEmpty(response.errors())) {
- throw new HoodieGlueSyncException("Fail to update partitions to " + tableId(databaseName, tableName)
- + " with error(s): " + response.errors());
- }
+ BatchUpdatePartitionResponse response = future.get();
+ if (CollectionUtils.nonEmpty(response.errors())) {
+ throw new HoodieGlueSyncException("Fail to update partitions to " + tableId(databaseName, table.name())
+ + " with error(s): " + response.errors());
}
} catch (Exception e) {
- throw new HoodieGlueSyncException("Fail to update partitions to " + tableId(databaseName, tableName), e);
+ throw new HoodieGlueSyncException("Fail to update partitions to " + tableId(databaseName, table.name()), e);
}
}
@Override
public void dropPartitions(String tableName, List<String> partitionsToDrop) {
- if (CollectionUtils.isNullOrEmpty(partitionsToDrop)) {
- LOG.info("No partitions to drop for " + tableName);
- return;
- }
- LOG.info("Drop " + partitionsToDrop.size() + "partition(s) in table " + tableId(databaseName, tableName));
+ HoodieTimer timer = HoodieTimer.start();
try {
- List<CompletableFuture<BatchDeletePartitionResponse>> futures = new ArrayList<>();
- for (List<String> batch : CollectionUtils.batches(partitionsToDrop, MAX_DELETE_PARTITIONS_PER_REQUEST)) {
+ if (partitionsToDrop.isEmpty()) {
+ LOG.info("No partitions to drop for " + tableId(this.databaseName, tableName));
+ return;
+ }
+ parallelizeChange(partitionsToDrop, this.changeParallelism, partitions -> this.dropPartitionsInternal(tableName, partitions), MAX_DELETE_PARTITIONS_PER_REQUEST);
+ } finally {
+ LOG.info("Deleted {} partitions to table {} in {} ms", partitionsToDrop.size(), tableId(this.databaseName, tableName), timer.endTimer());
+ }
+ }
- List<PartitionValueList> partitionValueLists = batch.stream().map(partition -> {
- PartitionValueList partitionValueList = PartitionValueList.builder()
- .values(partitionValueExtractor.extractPartitionValuesInPath(partition))
- .build();
- return partitionValueList;
- }).collect(Collectors.toList());
+ private void dropPartitionsInternal(String tableName, List<String> partitionsToDrop) {
+ try {
+ List<PartitionValueList> partitionValueLists = partitionsToDrop.stream().map(partition -> PartitionValueList.builder()
+ .values(partitionValueExtractor.extractPartitionValuesInPath(partition))
+ .build()
+ ).collect(Collectors.toList());
- BatchDeletePartitionRequest batchDeletePartitionRequest = BatchDeletePartitionRequest.builder()
- .databaseName(databaseName)
- .tableName(tableName)
- .partitionsToDelete(partitionValueLists)
- .build();
- futures.add(awsGlue.batchDeletePartition(batchDeletePartitionRequest));
- }
+ BatchDeletePartitionRequest batchDeletePartitionRequest = BatchDeletePartitionRequest.builder()
+ .databaseName(databaseName)
+ .tableName(tableName)
+ .partitionsToDelete(partitionValueLists)
+ .build();
+ CompletableFuture<BatchDeletePartitionResponse> future = awsGlue.batchDeletePartition(batchDeletePartitionRequest);
- for (CompletableFuture<BatchDeletePartitionResponse> future : futures) {
- BatchDeletePartitionResponse response = future.get();
- if (CollectionUtils.nonEmpty(response.errors())) {
- throw new HoodieGlueSyncException("Fail to drop partitions to " + tableId(databaseName, tableName)
- + " with error(s): " + response.errors());
- }
+ BatchDeletePartitionResponse response = future.get();
+ if (CollectionUtils.nonEmpty(response.errors())) {
+ throw new HoodieGlueSyncException("Fail to drop partitions to " + tableId(databaseName, tableName)
+ + " with error(s): " + response.errors());
}
} catch (Exception e) {
throw new HoodieGlueSyncException("Fail to drop partitions to " + tableId(databaseName, tableName), e);
diff --git a/hudi-aws/src/main/java/org/apache/hudi/config/GlueCatalogSyncClientConfig.java b/hudi-aws/src/main/java/org/apache/hudi/config/GlueCatalogSyncClientConfig.java
index 21244e65154..0f6ac76a166 100644
--- a/hudi-aws/src/main/java/org/apache/hudi/config/GlueCatalogSyncClientConfig.java
+++ b/hudi-aws/src/main/java/org/apache/hudi/config/GlueCatalogSyncClientConfig.java
@@ -26,6 +26,8 @@ import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
+import java.util.stream.IntStream;
+
/**
* Hoodie Configs for Glue.
*/
@@ -43,6 +45,28 @@ public class GlueCatalogSyncClientConfig extends HoodieConfig {
.sinceVersion("0.14.0")
.withDocumentation("Glue catalog sync based client will skip archiving the table version if this config is set to true");
+ public static final ConfigProperty<Integer> ALL_PARTITIONS_READ_PARALLELISM = ConfigProperty
+ .key(GLUE_CLIENT_PROPERTY_PREFIX + "all_partitions_read_parallelism")
+ .defaultValue(1)
+ .markAdvanced()
+ .withValidValues(IntStream.rangeClosed(1, 10).mapToObj(Integer::toString).toArray(String[]::new))
+ .sinceVersion("1.0.0")
+ .withDocumentation("Parallelism for listing all partitions(first time sync). Should be in interval [1, 10].");
+
+ public static final ConfigProperty<Integer> CHANGED_PARTITIONS_READ_PARALLELISM = ConfigProperty
+ .key(GLUE_CLIENT_PROPERTY_PREFIX + "changed_partitions_read_parallelism")
+ .defaultValue(1)
+ .markAdvanced()
+ .sinceVersion("1.0.0")
+ .withDocumentation("Parallelism for listing changed partitions(second and subsequent syncs).");
+
+ public static final ConfigProperty<Integer> PARTITION_CHANGE_PARALLELISM = ConfigProperty
+ .key(GLUE_CLIENT_PROPERTY_PREFIX + "partition_change_parallelism")
+ .defaultValue(1)
+ .markAdvanced()
+ .sinceVersion("1.0.0")
+ .withDocumentation("Parallelism for change operations - such as create/update/delete.");
+
public static final ConfigProperty<Boolean> GLUE_METADATA_FILE_LISTING = ConfigProperty
.key(GLUE_CLIENT_PROPERTY_PREFIX + "metadata_file_listing")
.defaultValue(false)
diff --git a/hudi-aws/src/test/java/org/apache/hudi/aws/sync/ITTestGluePartitionPushdown.java b/hudi-aws/src/test/java/org/apache/hudi/aws/sync/ITTestGluePartitionPushdown.java
index d9191fd5441..9601482b65a 100644
--- a/hudi-aws/src/test/java/org/apache/hudi/aws/sync/ITTestGluePartitionPushdown.java
+++ b/hudi-aws/src/test/java/org/apache/hudi/aws/sync/ITTestGluePartitionPushdown.java
@@ -48,12 +48,10 @@ import software.amazon.awssdk.services.glue.model.TableInput;
import java.io.IOException;
import java.nio.file.Files;
import java.time.Instant;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutionException;
-import static org.apache.hudi.hive.HiveSyncConfig.HIVE_SYNC_FILTER_PUSHDOWN_MAX_SIZE;
import static org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_BASE_PATH;
import static org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_DATABASE_NAME;
@@ -127,35 +125,14 @@ public class ITTestGluePartitionPushdown {
@Test
public void testEmptyPartitionShouldReturnEmpty() {
- Assertions.assertEquals(0, glueSync.getPartitionsByFilter(TABLE_NAME,
- glueSync.generatePushDownFilter(Arrays.asList("1/bar"), partitionsFieldSchema)).size());
+ Assertions.assertEquals(0, glueSync.getPartitionsFromList(TABLE_NAME,
+ Arrays.asList("1/bar")).size());
}
@Test
public void testPresentPartitionShouldReturnIt() throws ExecutionException, InterruptedException {
createPartitions("1", "b'ar");
- Assertions.assertEquals(1, glueSync.getPartitionsByFilter(TABLE_NAME,
- glueSync.generatePushDownFilter(Arrays.asList("1/b'ar", "2/foo", "1/b''ar"), partitionsFieldSchema)).size());
- }
-
- @Test
- public void testPresentPartitionShouldReturnAllWhenExpressionFilterLengthTooLong() throws ExecutionException, InterruptedException {
- createPartitions("1", "b'ar");
-
- // this will generate an expression larger than GLUE_EXPRESSION_MAX_CHARS
- List<String> tooLargePartitionPredicate = new ArrayList<>();
- for (int i = 0; i < 500; i++) {
- tooLargePartitionPredicate.add(i + "/foo");
- }
- Assertions.assertEquals(1, glueSync.getPartitionsByFilter(TABLE_NAME,
- glueSync.generatePushDownFilter(tooLargePartitionPredicate, partitionsFieldSchema)).size(),
- "Should fallback to listing all existing partitions");
-
- // now set the pushdown max size to a low value to transform the expression in lower/upper bound
- hiveSyncProps.setProperty(HIVE_SYNC_FILTER_PUSHDOWN_MAX_SIZE.key(), "10");
- glueSync = new AWSGlueCatalogSyncClient(new HiveSyncConfig(hiveSyncProps));
- Assertions.assertEquals(0, glueSync.getPartitionsByFilter(TABLE_NAME,
- glueSync.generatePushDownFilter(tooLargePartitionPredicate, partitionsFieldSchema)).size(),
- "No partitions should match");
+ Assertions.assertEquals(1, glueSync.getPartitionsFromList(TABLE_NAME,
+ Arrays.asList("1/b'ar", "2/foo", "1/b''ar")).size());
}
}
diff --git a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
index b0fb3098c10..ddc6da22d91 100644
--- a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
+++ b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
@@ -39,7 +39,6 @@ import org.apache.parquet.schema.MessageType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -383,18 +382,7 @@ public class HiveSyncTool extends HoodieSyncTool implements AutoCloseable {
return syncClient.getAllPartitions(tableName);
}
- List<String> partitionKeys = config.getSplitStrings(META_SYNC_PARTITION_FIELDS).stream()
- .map(String::toLowerCase)
- .collect(Collectors.toList());
-
- List<FieldSchema> partitionFields = syncClient.getMetastoreFieldSchemas(tableName)
- .stream()
- .filter(f -> partitionKeys.contains(f.getName()))
- .sorted(Comparator.comparing(f -> partitionKeys.indexOf(f.getName())))
- .collect(Collectors.toList());
-
- return syncClient.getPartitionsByFilter(tableName,
- syncClient.generatePushDownFilter(writtenPartitions, partitionFields));
+ return syncClient.getPartitionsFromList(tableName, writtenPartitions);
}
/**
diff --git a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
index 757d6028585..d3ef86a30a3 100644
--- a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
+++ b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HoodieHiveSyncClient.java
@@ -66,6 +66,7 @@ import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_USE_PRE_APACHE_INPU
import static org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_BASE_FILE_FORMAT;
import static org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_BASE_PATH;
import static org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_DATABASE_NAME;
+import static org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_PARTITION_FIELDS;
import static org.apache.hudi.sync.common.util.TableUtils.tableId;
/**
@@ -217,8 +218,19 @@ public class HoodieHiveSyncClient extends HoodieSyncClient {
}
@Override
- public List<Partition> getPartitionsByFilter(String tableName, String filter) {
+ public List<Partition> getPartitionsFromList(String tableName, List<String> partitions) {
+ String filter = null;
try {
+ List<String> partitionKeys = config.getSplitStrings(META_SYNC_PARTITION_FIELDS).stream()
+ .map(String::toLowerCase)
+ .collect(Collectors.toList());
+
+ List<FieldSchema> partitionFields = this.getMetastoreFieldSchemas(tableName)
+ .stream()
+ .filter(f -> partitionKeys.contains(f.getName()))
+ .collect(Collectors.toList());
+ filter = this.generatePushDownFilter(partitions, partitionFields);
+
return client.listPartitionsByFilter(databaseName, tableName, filter, (short)-1)
.stream()
.map(p -> new Partition(p.getValues(), p.getSd().getLocation()))
diff --git a/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieMetaSyncOperations.java b/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieMetaSyncOperations.java
index b1acaf14396..ca0bec3604b 100644
--- a/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieMetaSyncOperations.java
+++ b/hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/HoodieMetaSyncOperations.java
@@ -99,11 +99,9 @@ public interface HoodieMetaSyncOperations {
}
/**
- * Get the metadata of partitions that belong to the specified table
- * @param tableName
- * @return
+ * Get partitions given input list of partitions.
*/
- default List<Partition> getPartitionsByFilter(String tableName, String filter) {
+ default List<Partition> getPartitionsFromList(String tableName, List<String> partitionList) {
return Collections.emptyList();
}