wombatu-kun commented on code in PR #20244:
URL: https://github.com/apache/hudi/pull/20244#discussion_r4384588035
##########
hudi-client/hudi-java-client/src/main/java/org/apache/hudi/metadata/JavaHoodieMetadataBulkInsertPartitioner.java:
##########
@@ -29,19 +32,19 @@
* A {@code BulkInsertPartitioner} implementation for Metadata Table to
improve performance of initialization of metadata
* table partition when a very large number of records are inserted.
*
+ * <p>The records of a metadata table partition are tagged with the file group
they hash to. The partitioner sorts them
+ * by file group and then by record key, so each file group's records form one
sorted run that is written with that
+ * file group's file id prefix.
+ *
* @param <T> HoodieRecordPayload type
*/
public class JavaHoodieMetadataBulkInsertPartitioner<T>
implements BulkInsertPartitioner<List<HoodieRecord<T>>> {
- private String fileId = null;
@Override
public List<HoodieRecord<T>> repartitionRecords(List<HoodieRecord<T>>
records, int outputPartitions) {
- if (records.isEmpty()) {
- return records;
- }
- records.sort(Comparator.comparing(record ->
record.getKey().getRecordKey(), StringUtils.UTF8_LEXICOGRAPHIC_COMPARATOR));
- fileId =
HoodieTableMetadataUtil.getFileGroupPrefix(records.get(0).getCurrentLocation().getFileId());
+ records.sort(Comparator.comparing((HoodieRecord<T> record) ->
getFileIdPrefix(record))
Review Comment:
The comparator calls `getFileIdPrefix` on both sides of every comparison,
and each call allocates a substring, so the sort now allocates O(n log n)
strings. Every record of a file group is tagged with the identical file id, so
comparing `getCurrentLocation().getFileId()` gives the same grouping with no
allocation - optional, not a blocker.
##########
hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaBulkInsertHelper.java:
##########
@@ -111,17 +111,31 @@ public List<WriteStatus> bulkInsert(List<HoodieRecord<T>>
inputRecords,
final List<HoodieRecord<T>> repartitionedRecords =
(List<HoodieRecord<T>>) partitioner.repartitionRecords(dedupedRecords,
targetParallelism);
- String fileIdPrefix;
+ List<WriteStatus> writeStatuses = new ArrayList<>();
if (partitioner instanceof JavaHoodieMetadataBulkInsertPartitioner) {
- fileIdPrefix = partitioner.getFileIdPfx(0);
- } else {
- FileIdPrefixProvider fileIdPrefixProvider = (FileIdPrefixProvider)
ReflectionUtils.loadClass(
- config.getFileIdPrefixProviderClassName(),
- config.getProps());
- fileIdPrefix = fileIdPrefixProvider.createFilePrefix("");
+ // Metadata table records are tagged with the file group they hash to,
and the partitioner sorts them by file
+ // group. Each file group's run of records is written with its own file
id prefix and handle factory, so readers
+ // that look a key up in the file group it hashes to find it.
+ int runStart = 0;
+ int runIndex = 0;
+ for (int i = 1; i <= repartitionedRecords.size(); i++) {
Review Comment:
+1
--
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]
