vinishjail97 commented on code in PR #835:
URL: https://github.com/apache/incubator-xtable/pull/835#discussion_r4158844640
##########
xtable-core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java:
##########
@@ -84,12 +92,62 @@ public class BaseFileUpdatesExtractor {
* @param partitionedDataFiles files grouped by partition to sync
* @param metaClient the meta client for the Hudi table with the latest
timeline state loaded
* @param commit The current commit started by the Hudi client
+ * @param useExternalFileGroupPrefix whether a trailing {@code bucket-N}
directory is a file-group
+ * prefix (true only for a Paimon source)
* @return The information needed to create a "replace" commit for the Hudi
table
*/
ReplaceMetadata extractSnapshotChanges(
List<PartitionFileGroup> partitionedDataFiles,
HoodieTableMetaClient metaClient,
- String commit) {
+ String commit,
+ boolean useExternalFileGroupPrefix) {
+ return withFileSystemView(
+ metaClient,
+ (fsView, metadataConfig) ->
+ extractFromFsView(
+ partitionedDataFiles,
+ commit,
+ fsView,
+ metaClient,
+ metadataConfig,
+ useExternalFileGroupPrefix));
+ }
+
+ /**
+ * Returns true when the table still has live file groups under a {@code
bucket-N} partition.
+ * XTable 0.4.0 registered Paimon files under {@code <partition>/bucket-N},
and the incremental
+ * sync now keys the same files by {@code <partition>}, so a snapshot sync
must replace those file
+ * groups first. The replaced file groups are no longer live, so the check
is false after that
+ * sync.
+ *
+ * @param metaClient the meta client for the Hudi table with the latest
timeline state loaded
+ * @return true if a file group exists under a partition path that ends in
{@code bucket-N}
+ */
+ boolean
hasFileGroupsInExternalFileGroupPrefixPartitions(HoodieTableMetaClient
metaClient) {
Review Comment:
Removed the check in 2bef501. A Paimon to Hudi target from 0.4.0 now fails
the incremental sync with a message to run one FULL sync, and the detection
reads only the last commit's extra metadata.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java:
##########
@@ -84,12 +92,62 @@ public class BaseFileUpdatesExtractor {
* @param partitionedDataFiles files grouped by partition to sync
* @param metaClient the meta client for the Hudi table with the latest
timeline state loaded
* @param commit The current commit started by the Hudi client
+ * @param useExternalFileGroupPrefix whether a trailing {@code bucket-N}
directory is a file-group
+ * prefix (true only for a Paimon source)
* @return The information needed to create a "replace" commit for the Hudi
table
*/
ReplaceMetadata extractSnapshotChanges(
List<PartitionFileGroup> partitionedDataFiles,
HoodieTableMetaClient metaClient,
- String commit) {
+ String commit,
+ boolean useExternalFileGroupPrefix) {
+ return withFileSystemView(
+ metaClient,
+ (fsView, metadataConfig) ->
+ extractFromFsView(
+ partitionedDataFiles,
+ commit,
+ fsView,
+ metaClient,
+ metadataConfig,
+ useExternalFileGroupPrefix));
+ }
+
+ /**
+ * Returns true when the table still has live file groups under a {@code
bucket-N} partition.
+ * XTable 0.4.0 registered Paimon files under {@code <partition>/bucket-N},
and the incremental
+ * sync now keys the same files by {@code <partition>}, so a snapshot sync
must replace those file
+ * groups first. The replaced file groups are no longer live, so the check
is false after that
+ * sync.
+ *
+ * @param metaClient the meta client for the Hudi table with the latest
timeline state loaded
+ * @return true if a file group exists under a partition path that ends in
{@code bucket-N}
+ */
+ boolean
hasFileGroupsInExternalFileGroupPrefixPartitions(HoodieTableMetaClient
metaClient) {
+ if (!metaClient.isTimelineNonEmpty()) {
+ return false;
+ }
+ return withFileSystemView(
+ metaClient,
+ (fsView, metadataConfig) ->
+ FSUtils.getAllPartitionPaths(engineContext, metaClient,
metadataConfig).stream()
+ .filter(
+ partitionPath ->
+ EXTERNAL_FILE_GROUP_PREFIX_PATTERN
+
.matcher(partitionPath.substring(partitionPath.lastIndexOf('/') + 1))
+ .matches())
+ .anyMatch(
+ partitionPath ->
+
fsView.getLatestBaseFiles(partitionPath).findAny().isPresent()));
+ }
+
+ @FunctionalInterface
+ private interface FileSystemViewFunction<T> {
+ T apply(SyncableFileSystemView fsView, HoodieMetadataConfig
metadataConfig);
+ }
+
+ private <T> T withFileSystemView(
+ HoodieTableMetaClient metaClient, FileSystemViewFunction<T> function) {
Review Comment:
Removed in 2bef501, because the bucket-N check was its only other caller.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java:
##########
@@ -75,6 +75,14 @@ public class BaseFileUpdatesExtractor {
private static final Pattern HUDI_BASE_FILE_PATTERN =
Pattern.compile(
"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}-[0-9]_[0-9a-fA-F-]+_[0-9]+\\.");
+ // Paimon lays files out as {@code <partition>/bucket-N/<file>}. Hudi treats
the trailing {@code
+ // bucket-N} directory as a file-group prefix within the partition rather
than as part of the
+ // partition path (see Hudi PR #17788). Detecting such a directory lets us
register the file under
+ // its true partition with the prefix encoded into the external-file marker,
so an unpartitioned
+ // Paimon table is not mistakenly read as partitioned by "bucket-N". Other
sources can have a
+ // partition value that matches this pattern, so callers enable it only for
a Paimon source.
+ private static final Pattern EXTERNAL_FILE_GROUP_PREFIX_PATTERN =
+ Pattern.compile("bucket-[0-9]+");
Review Comment:
Done in 2bef501. `InternalDataFile` has an optional `fileGroupPrefix`, which
the Paimon source sets to `bucket-N`, so the Hudi target no longer matches
directory names or checks the source format.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java:
##########
@@ -222,35 +282,87 @@ ReplaceMetadata extractFromFsView(
* @param internalFilesDiff the diff to apply to the Hudi table
* @param commit The current commit started by the Hudi client
* @param indexVersion the Hudi index version
+ * @param useExternalFileGroupPrefix whether a trailing {@code bucket-N}
directory is a file-group
+ * prefix (true only for a Paimon source)
* @return The information needed to create a "replace" commit for the Hudi
table
*/
ReplaceMetadata convertDiff(
@NonNull InternalFilesDiff internalFilesDiff,
@NonNull String commit,
- @NonNull HoodieIndexVersion indexVersion) {
+ @NonNull HoodieIndexVersion indexVersion,
+ boolean useExternalFileGroupPrefix) {
// For all removed files, group by partition and extract the file id
Map<String, List<String>> partitionToReplacedFileIds =
internalFilesDiff.dataFilesRemoved().stream()
.map(file -> new CachingPath(file.getPhysicalPath()))
.collect(
Collectors.groupingBy(
- path -> HudiPathUtils.getPartitionPath(tableBasePath,
path),
- Collectors.mapping(this::getFileId, Collectors.toList())));
+ path -> truePartitionPath(tableBasePath, path,
useExternalFileGroupPrefix),
+ Collectors.mapping(
+ path -> getFileId(path, useExternalFileGroupPrefix),
Collectors.toList())));
// For all added files, group by partition and extract the file id
List<WriteStatus> writeStatuses =
internalFilesDiff.dataFilesAdded().stream()
- .map(file -> toWriteStatus(tableBasePath, commit, file,
Optional.empty(), indexVersion))
+ .map(
+ file ->
+ toWriteStatus(
+ tableBasePath,
+ commit,
+ file,
+ Optional.empty(),
+ indexVersion,
+ useExternalFileGroupPrefix))
.collect(CustomCollectors.toList(internalFilesDiff.dataFilesAdded().size()));
return ReplaceMetadata.of(partitionToReplacedFileIds, writeStatuses);
}
- private String getFileId(Path filePath) {
+ private String getFileId(Path filePath, boolean useExternalFileGroupPrefix) {
String fileName = filePath.getName();
// if file was created by Hudi use original fileId, otherwise use the file
name as IDs
if (isFileCreatedByHudiWriter(fileName)) {
return FSUtils.getFileId(fileName);
}
- return fileName;
+ // External bucketed files keep their file-group prefix as part of the
fileId so the prefix can
+ // be recovered when Hudi resolves the physical path of the externally
created file.
+ return externalFileGroupPrefix(filePath, useExternalFileGroupPrefix)
+ .map(prefix -> prefix + "/" + fileName)
+ .orElse(fileName);
+ }
+
+ /**
+ * Returns the external file-group prefix (e.g. Paimon's {@code bucket-N})
when the file's
+ * immediate parent directory denotes a file group within the partition
rather than a partition
+ * segment, otherwise empty. Always empty when {@code
useExternalFileGroupPrefix} is false.
+ */
+ private Optional<String> externalFileGroupPrefix(
+ Path filePath, boolean useExternalFileGroupPrefix) {
+ if (!useExternalFileGroupPrefix) {
+ return Optional.empty();
+ }
+ Path parent = filePath.getParent();
+ if (parent == null) {
+ return Optional.empty();
+ }
+ String parentName = parent.getName();
+ return EXTERNAL_FILE_GROUP_PREFIX_PATTERN.matcher(parentName).matches()
+ ? Optional.of(parentName)
+ : Optional.empty();
+ }
+
+ /**
+ * Resolves the true Hudi partition path for a file, stripping any trailing
external file-group
+ * prefix directory (e.g. {@code bucket-N}) so it is not treated as part of
the partition.
+ */
+ private String truePartitionPath(
+ Path tableBasePath, Path filePath, boolean useExternalFileGroupPrefix) {
+ String partitionPath = HudiPathUtils.getPartitionPath(tableBasePath,
filePath);
+ Optional<String> prefix = externalFileGroupPrefix(filePath,
useExternalFileGroupPrefix);
+ if (!prefix.isPresent()) {
+ return partitionPath;
+ }
+ return partitionPath.equals(prefix.get())
+ ? ""
+ : partitionPath.substring(0, partitionPath.length() -
prefix.get().length() - 1);
Review Comment:
Done in 2bef501.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java:
##########
@@ -148,9 +155,13 @@ public CommitsBacklog<HoodieInstant> getCommitsBacklog(
CommitsPair lastPendingHoodieInstantsCommitsPair =
getCompletedAndPendingCommitsForInstants(lastPendingInstants);
List<HoodieInstant> commitsToProcessNext =
- mergeAndDedupLists(
- lastPendingHoodieInstantsCommitsPair.getCompletedCommits(),
- commitsPair.getCompletedCommits());
+ usesCompletionTimeOrdering()
+ ? orderByCompletionTimeAndDedup(
+ lastPendingHoodieInstantsCommitsPair.getCompletedCommits(),
+ commitsPair.getCompletedCommits())
+ : mergeAndDedupLists(
+ lastPendingHoodieInstantsCommitsPair.getCompletedCommits(),
+ commitsPair.getCompletedCommits());
Review Comment:
Done in 2bef501. `mergeAndDedupLists` now takes a key and an ordering, and
it covers both the completed commits and the pending instants.
--
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]
