vinishjail97 commented on code in PR #835:
URL: https://github.com/apache/incubator-xtable/pull/835#discussion_r4108765672
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java:
##########
@@ -237,10 +248,86 @@ private HoodieTimeline getCompletedCommits() {
return metaClient.getActiveTimeline().filterCompletedInstants();
}
+ /**
+ * Table version 8+ (Hudi 1.x, timeline layout V2) makes a commit visible at
its completion time
+ * rather than its requested (instant) time, so incremental selection and
ordering must be based
+ * on completion time. Table version 6 keeps the legacy requested-time
ordering.
+ */
+ private boolean usesCompletionTimeOrdering() {
+ return metaClient
+ .getTableConfig()
+ .getTableVersion()
+ .greaterThanOrEquals(HoodieTableVersion.EIGHT);
+ }
+
+ private HoodieInstant getLatestCompletedInstant(HoodieTimeline
completedTimeline) {
+ Option<HoodieInstant> latestCommit =
+ usesCompletionTimeOrdering()
+ ? Option.fromJavaOptional(
+ completedTimeline
+ .getInstantsOrderedByCompletionTime()
+ .reduce((first, second) -> second))
+ : completedTimeline.lastInstant();
+ return latestCommit.orElseThrow(
+ () -> new ReadException("Unable to read latest commit from Hudi source
table"));
+ }
+
+ /**
+ * Selects the commits that completed after the last synced commit's
completion time, ordered by
+ * completion time. Unlike the requested-time path this also surfaces
commits whose requested time
+ * is older than the last synced commit but whose completion is newer
(out-of-order completion).
+ */
+ private CommitsPair getCompletedAndPendingCommitsAfterCompletionTime(
+ HoodieInstant commitInstant) {
+ List<HoodieInstant> modifiedAfter =
+ metaClient
+ .getActiveTimeline()
+
.findInstantsModifiedAfterByCompletionTime(commitInstant.getCompletionTime())
+ .getInstants();
+ List<HoodieInstant> completedInstants =
+ modifiedAfter.stream()
+ .filter(HoodieInstant::isCompleted)
+ .sorted(Comparator.comparing(HoodieInstant::getCompletionTime))
+ .collect(Collectors.toList());
+ List<Instant> pendingInstants =
+ modifiedAfter.stream()
+ .filter(hoodieInstant -> hoodieInstant.isInflight() ||
hoodieInstant.isRequested())
+ .map(
+ hoodieInstant ->
+
HudiInstantUtils.parseFromInstantTime(hoodieInstant.requestedTime()))
+ .collect(Collectors.toList());
+ return CommitsPair.builder()
+ .completedCommits(completedInstants)
+ .pendingCommits(pendingInstants)
+ .build();
+ }
+
+ /**
+ * Merges two completed-commit lists, dedupes by requested time and action,
and orders by
+ * completion time. The action is part of the dedup key because distinct
actions can legally share
+ * a requested time: a savepoint instant reuses the requested time of the
commit it pins, and
+ * keying on requested time alone would drop it from the backlog.
+ */
+ private List<HoodieInstant> orderByCompletionTimeAndDedup(
Review Comment:
Fixed in 7ceff18. On table version 9 the checkpoint is now the completion
time, and `testLaggingTargetWithOutOfOrderCompletedCommits` covers the repro on
v6 and v9.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java:
##########
@@ -273,17 +315,31 @@ private WriteStatus toWriteStatus(
WriteStatus writeStatus = new WriteStatus();
Path path = new CachingPath(file.getPhysicalPath());
String partitionPath =
- partitionPathOptional.orElseGet(() ->
HudiPathUtils.getPartitionPath(tableBasePath, path));
+ partitionPathOptional.orElseGet(() -> truePartitionPath(tableBasePath,
path));
String fileId = getFileId(path);
String filePath =
path.toUri().getPath().substring(tableBasePath.toUri().getPath().length() + 1);
String fileName = path.getName();
+ Optional<String> fileGroupPrefix = externalFileGroupPrefix(path);
+ // For external bucketed files encode the file-group prefix in the marker
and keep the file name
+ // (not the bucket-relative path) as the marked name; otherwise fall back
to the plain marker on
+ // the full relative path. In both cases the directory portion is
preserved as-is.
+ String markedPath =
+ fileGroupPrefix
+ .map(
+ prefix ->
+ filePath.substring(0, filePath.length() -
fileName.length())
+ +
ExternalFilePathUtil.appendCommitTimeAndExternalFileMarker(
+ fileName, commitTime, prefix))
Review Comment:
Kept the prefix for v6 targets, because without it an unpartitioned Paimon
table is read as partitioned by `bucket-N`. The release notes will say that
Paimon to Hudi targets need a Hudi 1.x reader.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java:
##########
@@ -250,7 +257,42 @@ private String getFileId(Path filePath) {
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)
+ .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.
+ */
+ private Optional<String> externalFileGroupPrefix(Path filePath) {
+ 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) {
Review Comment:
Fixed in 7ceff18. The new `ConversionTarget#isIncrementalSyncSafe` returns
false when a Paimon-sourced Hudi target still has file groups under `bucket-N`
partitions, so one snapshot sync replaces them.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java:
##########
@@ -75,6 +75,13 @@ 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]+\\.");
+ // External bucketed sources (e.g. Paimon) lay 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 external table is not mistakenly read as
partitioned by "bucket-N".
+ private static final Pattern EXTERNAL_FILE_GROUP_PREFIX_PATTERN =
Review Comment:
Fixed in 7ceff18. The prefix now applies only when the source table format
is Paimon.
##########
xtable-core/src/test/java/org/apache/xtable/ITConversionController.java:
##########
@@ -153,7 +155,16 @@ private static Stream<Arguments>
generateTestParametersForFormatsSyncModesAndPar
for (String sourceFormat : Arrays.asList(HUDI, DELTA, ICEBERG, PAIMON)) {
for (SyncMode syncMode : SyncMode.values()) {
for (boolean isPartitioned : new boolean[] {true, false}) {
- arguments.add(Arguments.of(sourceFormat, syncMode, isPartitioned));
+ if (sourceFormat.equals(HUDI)) {
+ // Hudi is the source here (not a target), so the Hudi target
version does not apply.
Review Comment:
Added in 7ceff18. `testConcurrentInsertWritesInSource` now runs on v6 and
v9, and the new tests cover an incremental sync with an inflight commit and the
lagging target.
##########
xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSource.java:
##########
@@ -337,9 +431,9 @@ public void testForIncrementalSyncSafetyCheck() {
HoodieTableType tableType = HoodieTableType.COPY_ON_WRITE;
HudiTestUtil.PartitionConfig partitionConfig =
HudiTestUtil.PartitionConfig.of(null, null);
String tableName = GenericTable.getTableName();
- try (TestJavaHudiTable table =
- TestJavaHudiTable.forStandardSchema(
- tableName, tempDir, partitionConfig.getHudiConfig(), tableType)) {
+ try (TestSparkHudiTable table =
Review Comment:
Done in 7ceff18. Both tests now run on SIX and NINE, and the v9 safety check
also uses completion-time checkpoints.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java:
##########
@@ -288,7 +296,9 @@ public void syncFilesForDiff(InternalFilesDiff
internalFilesDiff) {
public void beginSync(InternalTable table) {
if (!metaClient.isPresent()) {
metaClient =
- Optional.of(hudiTableManager.initializeHudiTable(tableDataPath,
table, databaseName));
+ Optional.of(
+ hudiTableManager.initializeHudiTable(
+ tableDataPath, table, databaseName, tableVersion));
Review Comment:
Added in 7ceff18. The warning includes the base path, the existing version,
and the configured version.
##########
xtable-core/src/main/java/org/apache/xtable/hudi/HudiTargetConfig.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hudi;
+
+import java.util.Properties;
+
+import lombok.Value;
+
+import org.apache.hudi.common.table.HoodieTableVersion;
+
+/** Configuration of the Hudi conversion target. */
+@Value
+public class HudiTargetConfig {
+ /**
+ * Table format version to write for the Hudi target. Supported values are
{@code 6} (the legacy
+ * 0.x timeline layout, column-stats index V1) and {@code 9} (the Hudi 1.x
timeline layout,
+ * column-stats index V2). Defaults to {@code 9}.
Review Comment:
Fixed in 7ceff18. The comments now say the default is 6, and a non-numeric
or unknown value throws `IllegalArgumentException` with the config key.
--
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]
