vinishjail97 commented on code in PR #835:
URL: https://github.com/apache/incubator-xtable/pull/835#discussion_r4108273843
##########
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:
This writes the `_fg%3D<prefix>_hudiext` marker for v6 targets as well. Hudi
0.15's `ExternalFilePathUtil` only strips the last `_<segment>`, so a 0.x
reader would look for `data.parquet_<commitTime>` in the wrong directory. Would
it be reasonable to use the prefix only when the table version is NINE, since
the PR keeps v6 as the default for 0.x readers?
##########
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:
This pattern applies to every source format, not only Paimon. A
non-hive-style partition value such as `bucket-3` (for example, a Hudi-style
layout read through the Iceberg or Delta source with a custom location) would
lose its partition segment. Would it be reasonable to enable this only for the
Paimon source, for example through a flag on the target's source-format context?
##########
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:
Every Hudi source table in this class is written at v6, so the v9
completion-time source path is never exercised through `ConversionController`,
where `lastInstantSynced` and the pending list round-trip through target
metadata. The `ITHudiConversionSource` out-of-order test calls
`getCommitsBacklog` with a synthetic checkpoint that a real run does not
produce. Would it be reasonable to add v9-source cases here? I have three
locally (snapshot with an inflight commit, incremental with an inflight commit,
and the lagging-target case).
##########
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:
`testForIncrementalSyncSafetyCheck` and `testOnlyUpsertsAfterInserts` still
run at v6 only, so `isAffectedByCleanupProcess` has no v9 coverage. Would it
work to parameterize them over SIX and NINE like the other cases?
##########
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:
On a Paimon to Hudi target created by 0.4.0, files live in partition
`dt=x/bucket-0` with fileId `<fileName>`. After this change, incremental
removals are keyed as partition `dt=x` and fileId `bucket-0/<fileName>`, so I
am not sure the replace commit matches the old file groups, and compacted-away
files could stay visible. A snapshot sync looks self-healing through
`partitionPathsToDrop`. Would it work to force a snapshot sync when the
existing target still has `bucket-N` partitions, or to note this in the upgrade
docs?
##########
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:
With a completion-ordered backlog, a target that syncs `[late, early]` in
one run records `early`'s requested time as `lastInstantSynced`. When another
target lags, the next run replays `[late, early]`, and
`TableFormatSync.isChangeApplicableForLastSyncMetadata` re-applies `late`
because its requested time is after `early`. A local `ITConversionController`
test (sync Iceberg alone through both commits, then sync Iceberg and Delta)
gives Iceberg 200 rows against 150 in the source on v9, and passes on v6. Would
using completion time as the v9 sync checkpoint work here? I think this blocks
enabling `table_version=9` sources for multi-target syncs, and I can push the
repro test.
##########
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:
The configured version applies only when the table is created. An existing
v6 table with `table_version=9` stays at v6 without a message. A WARN with the
base path when the two differ would help users. This is a suggestion and does
not block.
##########
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:
Nits: this Javadoc and the comments at `HudiConversionTarget.java:117`,
`HudiConversionTarget.java:641` and `HudiTableManager.java:127` still say the
default is 9. On line 46, a value like `abc` throws a raw
`NumberFormatException`, and an unknown code can fail in `fromVersionCode`
before the clearer message below, so a single try/catch that rethrows the
`IllegalArgumentException` would cover both.
--
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]
