vinishjail97 commented on code in PR #835:
URL: https://github.com/apache/incubator-xtable/pull/835#discussion_r4187228353
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java:
##########
@@ -302,39 +407,24 @@ private List<HoodieInstant>
getCommitsForInstants(List<Instant> instants) {
}
/**
- * Merges commits from two lists and returns new list of sorted commits by
eliminating duplicates.
- *
- * @param list1 First sorted input list of commits
- * @param list2 Second sorted input list of commits.
- * @return merged list of commits in sorted order.
+ * Merges two lists, keeps the first element for each key, and sorts the
result. Commits are
+ * sorted with the timeline layout's ordering, which is the requested time
on table version 6 and
+ * the completion time on version 9. Commits are keyed by requested time and
action, because a
+ * savepoint instant reuses the requested time of the commit it pins.
*/
- private <T extends Comparable<T>> List<T> mergeAndDedupLists(
- @NonNull List<T> list1, @NonNull List<T> list2) {
- List<T> mergedList = new ArrayList<>();
- PeekingIterator<T> itr1 = Iterators.peekingIterator(list1.iterator());
- PeekingIterator<T> itr2 = Iterators.peekingIterator(list2.iterator());
- while (itr1.hasNext() || itr2.hasNext()) {
- if (!itr2.hasNext()) {
- mergedList.add(itr1.next());
- } else if (!itr1.hasNext()) {
- mergedList.add(itr2.next());
- } else {
- T element1 = itr1.peek();
- T element2 = itr2.peek();
- if (element1.compareTo(element2) < 0) {
- mergedList.add(element1);
- itr1.next();
- } else if (element1.compareTo(element2) > 0) {
- mergedList.add(element2);
- itr2.next();
- } else {
- mergedList.add(element1);
- itr1.next();
- itr2.next();
- }
- }
- }
- return mergedList;
+ private <T, K> List<T> mergeAndDedupLists(
+ @NonNull List<T> list1,
+ @NonNull List<T> list2,
+ Function<T, K> keyExtractor,
+ Comparator<T> ordering) {
+ Map<K, T> dedupedByKey = new LinkedHashMap<>();
Review Comment:
The result is sorted after the dedup, so insertion order is not needed.
Replaced it with `Collectors.toMap` in 2385f72.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java:
##########
@@ -269,26 +306,75 @@ public void syncFilesForSnapshot(List<PartitionFileGroup>
partitionedDataFiles)
baseFileUpdatesExtractor.extractSnapshotChanges(
partitionedDataFiles, getMetaClient(),
commitState.getInstantTime());
commitState.setReplaceMetadata(replaceMetadata);
+ // A snapshot sync replaces every file group, so the table uses the
file-group prefix layout.
+ commitState.setFileGroupPrefixLayout(true);
}
@Override
public void syncFilesForDiff(InternalFilesDiff internalFilesDiff) {
if (!metaClient.isPresent()) {
throw new IllegalStateException("Meta client is not initialized");
}
+ boolean fileGroupPrefixLayout =
lastCommitHasFileGroupPrefixLayout(metaClient.get());
+ if (!fileGroupPrefixLayout && hasPartitionSubdirectory(internalFilesDiff))
{
+ throw new NotSupportedException(
+ String.format(
+ "Hudi target table at %s was written from a Paimon source by an
XTable version earlier"
+ + " than 0.5.0. That version registered the files under
<partition>/bucket-N"
+ + " partitions, and an incremental sync cannot update this
layout. Run one sync"
+ + " with syncMode FULL, and incremental sync then resumes.",
+ tableDataPath));
+ }
HoodieIndexVersion indexVersion =
existingIndexVersionOrDefault(PARTITION_NAME_COLUMN_STATS,
metaClient.get());
BaseFileUpdatesExtractor.ReplaceMetadata replaceMetadata =
baseFileUpdatesExtractor.convertDiff(
internalFilesDiff, commitState.getInstantTime(), indexVersion);
commitState.setReplaceMetadata(replaceMetadata);
+ // An incremental sync keeps the file groups in place, so it carries the
layout forward.
+ commitState.setFileGroupPrefixLayout(fileGroupPrefixLayout);
+ }
+
+ private static boolean hasPartitionSubdirectory(InternalFilesDiff
internalFilesDiff) {
+ return Stream.concat(
+ internalFilesDiff.dataFilesAdded().stream(),
+ internalFilesDiff.dataFilesRemoved().stream())
+ .anyMatch(file -> file.getPartitionSubdirectory().isPresent());
+ }
+
+ /**
+ * XTable 0.4.0 registered Paimon files under {@code <partition>/bucket-N}
partitions and did not
Review Comment:
Updated the Javadoc and the exception message in 2385f72 to describe the
layout without naming a format or version.
--
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]
