the-other-tim-brown commented on code in PR #835:
URL: https://github.com/apache/incubator-xtable/pull/835#discussion_r4116414511
##########
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:
Is there a more general way that we can do this? Something like Iceberg does
not have strict requirements that all paths be under the same source for
example so it makes sense to find a more general way to do this in my opinion.
##########
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:
Nitpick: My preference is to use the functional style here like
```
return prefix.map(p -> p.equals(partitionPath) ? "" :
partitionPath.substring(0, partitionPath.length() - p.length() -
1)).orElse(partitionPath);
```
##########
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:
Nitpick: my preference would be to keep all the logic under a single method,
`mergeAndDedupLists` or some renamed version of this method that better
captures what is happening.
##########
xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiConversionSource.java:
##########
@@ -71,6 +72,7 @@ void
testIsIncrementalSyncSafeFromWithNullEarliestCommitToRetainNoCleanInstants(
when(mockMetaClient.getActiveTimeline()).thenReturn(mockActiveTimeline);
when(mockMetaClient.getTableConfig()).thenReturn(mockTableConfig);
when(mockTableConfig.isMetadataTableAvailable()).thenReturn(false);
+ when(mockTableConfig.getTableVersion()).thenReturn(HoodieTableVersion.SIX);
Review Comment:
Should we update any tests to return version 9 instead to get better
coverage?
##########
xtable-core/src/test/java/org/apache/xtable/hudi/TestBaseFileUpdatesExtractor.java:
##########
@@ -145,7 +145,7 @@ void convertDiff() {
BaseFileUpdatesExtractor extractor =
BaseFileUpdatesExtractor.of(CONTEXT, new CachingPath(tableBasePath));
BaseFileUpdatesExtractor.ReplaceMetadata replaceMetadata =
- extractor.convertDiff(diff, COMMIT_TIME, HoodieIndexVersion.V1);
+ extractor.convertDiff(diff, COMMIT_TIME, HoodieIndexVersion.V1, false);
Review Comment:
Should we have anything with `true` for `useExternalFileGroupPrefix`?
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java:
##########
@@ -124,10 +124,17 @@ public InternalSnapshot getCurrentSnapshot() {
@Override
public TableChange getTableChangeForCommit(HoodieInstant
hoodieInstantForDiff) {
HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
+ // The set of commits visible as-of the diff commit is ordered by
completion time on table
+ // version 9 (timeline layout V2) and by requested time on version 6.
HoodieTimeline visibleTimeline =
- activeTimeline
- .filterCompletedInstants()
- .findInstantsBeforeOrEquals(hoodieInstantForDiff.requestedTime());
+ usesCompletionTimeOrdering()
Review Comment:
Are there any utils provided by Hudi that we could use to handle different
layout types instead of adding the cases in our own codebase?
##########
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:
Let's give this a better name? `function` is ambiguous and makes it a bit
unclear when to use this method
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java:
##########
@@ -648,14 +723,16 @@ private HoodieWriteConfig getWriteConfig(
HoodieMetadataConfig.newBuilder()
.enable(true)
.withProperties(properties)
- // Hudi 1.2.0 couples the partition-stats index to the
column-stats index. For
- // partitioned tables, the partition-stats generation path
rebuilds a file-system
- // view over the committed external parquet files and groups
them by fileId.
- // XTable's externally-registered files have non-Hudi names
whose fileId cannot be
- // parsed once the "_hudiext" marker is stripped, which
leads to failures. So
- // column stats are only enabled for un-partitioned tables
for now. Tracked in
- // https://github.com/apache/incubator-xtable/issues/832.
-
.withMetadataIndexColumnStats(!metaClient.getTableConfig().isTablePartitioned())
+ // Build the column-stats index for all tables. The
partition-stats index is
+ // disabled independently: its generation path rebuilds a
file-system view over
+ // the committed external parquet files and groups them by
fileId, but XTable's
+ // externally-registered files have non-Hudi names whose
fileId cannot be parsed
+ // once the "_hudiext" marker is stripped, which leads to
failures on partitioned
+ // tables. Disabling partition stats (while keeping column
stats) requires the
+ // independent toggle added in
https://github.com/apache/hudi/pull/19111. Tracked
Review Comment:
Can this comment be updated now? The linked PR is merged.
--
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]
