diff --git a/pom.xml b/pom.xml index 910d794b0..4e3d06f4c 100644 --- a/pom.xml +++ b/pom.xml @@ -70,7 +70,7 @@ 1.18.46 1.18.20.0 3.4.1 - 1.2.0 + 1.3.0-SNAPSHOT 2.29.40 3.1.3 3.3.1 diff --git a/xtable-core/src/main/java/org/apache/hudi/stats/XTableValueMetadata.java b/xtable-core/src/main/java/org/apache/hudi/metadata/stats/XTableValueMetadata.java similarity index 99% rename from xtable-core/src/main/java/org/apache/hudi/stats/XTableValueMetadata.java rename to xtable-core/src/main/java/org/apache/hudi/metadata/stats/XTableValueMetadata.java index ac6c70b21..6efa58f78 100644 --- a/xtable-core/src/main/java/org/apache/hudi/stats/XTableValueMetadata.java +++ b/xtable-core/src/main/java/org/apache/hudi/metadata/stats/XTableValueMetadata.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.stats; +package org.apache.hudi.metadata.stats; import static org.apache.xtable.model.schema.InternalSchema.MetadataKey.TIMESTAMP_PRECISION; import static org.apache.xtable.model.schema.InternalSchema.MetadataValue.MICROS; diff --git a/xtable-core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java b/xtable-core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java index 44acd089b..5cbe0261b 100644 --- a/xtable-core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java +++ b/xtable-core/src/main/java/org/apache/xtable/hudi/BaseFileUpdatesExtractor.java @@ -57,9 +57,9 @@ import org.apache.hudi.hadoop.fs.CachingPath; import org.apache.hudi.metadata.HoodieIndexVersion; import org.apache.hudi.metadata.HoodieTableMetadata; -import org.apache.hudi.stats.HoodieColumnRangeMetadata; -import org.apache.hudi.stats.ValueMetadata; -import org.apache.hudi.stats.XTableValueMetadata; +import org.apache.hudi.metadata.stats.HoodieColumnRangeMetadata; +import org.apache.hudi.metadata.stats.ValueMetadata; +import org.apache.hudi.metadata.stats.XTableValueMetadata; import org.apache.xtable.collectors.CustomCollectors; import org.apache.xtable.exception.ReadException; @@ -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 /bucket-N/}. + // 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 = + Pattern.compile("bucket-[0-9]+"); private final HoodieEngineContext engineContext; private final Path tableBasePath; @@ -234,7 +241,7 @@ ReplaceMetadata convertDiff( .map(file -> new CachingPath(file.getPhysicalPath())) .collect( Collectors.groupingBy( - path -> HudiPathUtils.getPartitionPath(tableBasePath, path), + path -> truePartitionPath(tableBasePath, path), Collectors.mapping(this::getFileId, Collectors.toList()))); // For all added files, group by partition and extract the file id List writeStatuses = @@ -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 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) { + String partitionPath = HudiPathUtils.getPartitionPath(tableBasePath, filePath); + Optional prefix = externalFileGroupPrefix(filePath); + if (!prefix.isPresent()) { + return partitionPath; + } + return partitionPath.equals(prefix.get()) + ? "" + : partitionPath.substring(0, partitionPath.length() - prefix.get().length() - 1); } /** @@ -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 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)) + .orElseGet( + () -> + ExternalFilePathUtil.appendCommitTimeAndExternalFileMarker( + filePath, commitTime)); writeStatus.setFileId(fileId); writeStatus.setPartitionPath(partitionPath); HoodieDeltaWriteStat writeStat = new HoodieDeltaWriteStat(); writeStat.setFileId(fileId); - writeStat.setPath( - ExternalFilePathUtil.appendCommitTimeAndExternalFileMarker(filePath, commitTime)); + writeStat.setPath(markedPath); writeStat.setPartitionPath(partitionPath); writeStat.setNumWrites(file.getRecordCount()); writeStat.setTotalWriteBytes(file.getFileSizeBytes()); @@ -338,7 +394,6 @@ private ReplaceMetadata combine(ReplaceMetadata other) { } private String getPartitionPath(Path tableBasePath, List files) { - return HudiPathUtils.getPartitionPath( - tableBasePath, new CachingPath(files.get(0).getPhysicalPath())); + return truePartitionPath(tableBasePath, new CachingPath(files.get(0).getPhysicalPath())); } } diff --git a/xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java index ecf3877e8..e0f5dd357 100644 --- a/xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java +++ b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java @@ -23,10 +23,13 @@ import java.time.Instant; import java.util.ArrayList; import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.stream.Collectors; +import java.util.stream.Stream; import lombok.Builder; import lombok.NonNull; @@ -35,6 +38,7 @@ import org.apache.hudi.avro.model.HoodieCleanMetadata; import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; import org.apache.hudi.common.table.timeline.HoodieActiveTimeline; import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.hudi.common.table.timeline.HoodieTimeline; @@ -83,13 +87,7 @@ public InternalTable getTable(HoodieInstant commit) { public InternalTable getCurrentTable() { HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline(); HoodieTimeline completedTimeline = activeTimeline.filterCompletedInstants(); - // get latest commit - HoodieInstant latestCommit = - completedTimeline - .lastInstant() - .orElseThrow( - () -> new ReadException("Unable to read latest commit from Hudi source table")); - return getTable(latestCommit); + return getTable(getLatestCompletedInstant(completedTimeline)); } @Override @@ -97,16 +95,18 @@ public InternalSnapshot getCurrentSnapshot() { HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline(); HoodieTimeline completedTimeline = activeTimeline.filterCompletedInstants(); // get latest commit - HoodieInstant latestCommit = - completedTimeline - .lastInstant() - .orElseThrow( - () -> new ReadException("Unable to read latest commit from Hudi source table")); + HoodieInstant latestCommit = getLatestCompletedInstant(completedTimeline); + // On table version 9 (timeline layout V2) a commit becomes visible at its completion time, so + // an instant with an earlier requested time may complete after the latest commit. Capture all + // currently inflight/requested instants as pending so none are missed; on version 6 keep the + // historical requested-time window. List pendingInstants = - activeTimeline - .filterInflightsAndRequested() - .findInstantsBefore(latestCommit.requestedTime()) - .getInstants(); + usesCompletionTimeOrdering() + ? activeTimeline.filterInflightsAndRequested().getInstants() + : activeTimeline + .filterInflightsAndRequested() + .findInstantsBefore(latestCommit.requestedTime()) + .getInstants(); InternalTable table = getTable(latestCommit); return InternalSnapshot.builder() .table(table) @@ -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() + ? activeTimeline + .filterCompletedInstants() + .findInstantsModifiedBeforeOrEqualsByCompletionTime( + hoodieInstantForDiff.getCompletionTime()) + : activeTimeline + .filterCompletedInstants() + .findInstantsBeforeOrEquals(hoodieInstantForDiff.requestedTime()); InternalTable table = getTable(hoodieInstantForDiff); return TableChange.builder() .tableAsOfChange(table) @@ -148,9 +155,13 @@ public CommitsBacklog getCommitsBacklog( CommitsPair lastPendingHoodieInstantsCommitsPair = getCompletedAndPendingCommitsForInstants(lastPendingInstants); List commitsToProcessNext = - mergeAndDedupLists( - lastPendingHoodieInstantsCommitsPair.getCompletedCommits(), - commitsPair.getCompletedCommits()); + usesCompletionTimeOrdering() + ? orderByCompletionTimeAndDedup( + lastPendingHoodieInstantsCommitsPair.getCompletedCommits(), + commitsPair.getCompletedCommits()) + : mergeAndDedupLists( + lastPendingHoodieInstantsCommitsPair.getCompletedCommits(), + commitsPair.getCompletedCommits()); List pendingInstantsToProcessNext = mergeAndDedupLists( lastPendingHoodieInstantsCommitsPair.getPendingCommits(), @@ -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 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 modifiedAfter = + metaClient + .getActiveTimeline() + .findInstantsModifiedAfterByCompletionTime(commitInstant.getCompletionTime()) + .getInstants(); + List completedInstants = + modifiedAfter.stream() + .filter(HoodieInstant::isCompleted) + .sorted(Comparator.comparing(HoodieInstant::getCompletionTime)) + .collect(Collectors.toList()); + List 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 orderByCompletionTimeAndDedup( + List list1, List list2) { + Map dedupedByRequestedTimeAndAction = new LinkedHashMap<>(); + Stream.concat(list1.stream(), list2.stream()) + .forEach( + hoodieInstant -> + dedupedByRequestedTimeAndAction.putIfAbsent( + hoodieInstant.requestedTime() + "_" + hoodieInstant.getAction(), + hoodieInstant)); + return dedupedByRequestedTimeAndAction.values().stream() + .sorted(Comparator.comparing(HoodieInstant::getCompletionTime)) + .collect(Collectors.toList()); + } + private CommitsPair getCompletedAndPendingCommitsAfterInstant(HoodieInstant commitInstant) { + if (usesCompletionTimeOrdering()) { + return getCompletedAndPendingCommitsAfterCompletionTime(commitInstant); + } // Table version 6 uses the old timeline view, so instants are selected and ordered by their - // requested (instant) time. Completion-time based handling will be added with table version 9 - // support in a follow-up PR. + // requested (instant) time. List allInstants = metaClient .getActiveTimeline() diff --git a/xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java index 84a1ffa0d..c8398a2bf 100644 --- a/xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java +++ b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java @@ -50,7 +50,7 @@ import org.apache.hudi.client.WriteStatus; import org.apache.hudi.client.common.HoodieJavaEngineContext; import org.apache.hudi.client.timeline.HoodieTimelineArchiver; -import org.apache.hudi.client.timeline.versioning.v1.TimelineArchiverV1; +import org.apache.hudi.client.timeline.TimelineArchivers; import org.apache.hudi.common.HoodieCleanStat; import org.apache.hudi.common.config.HoodieMetadataConfig; import org.apache.hudi.common.engine.HoodieEngineContext; @@ -80,6 +80,7 @@ import org.apache.hudi.metadata.HoodieIndexVersion; import org.apache.hudi.metadata.HoodieTableMetadataWriter; import org.apache.hudi.table.HoodieJavaTable; +import org.apache.hudi.table.HoodieTable; import org.apache.hudi.table.action.clean.CleanPlanner; import com.google.common.annotations.VisibleForTesting; @@ -112,6 +113,9 @@ public class HudiConversionTarget implements ConversionTarget { private CommitState commitState; // database to register the target table under, resolved from the target namespace private String databaseName; + // Hudi table format version to write (6 = legacy 0.x layout, 9 = Hudi 1.x layout), resolved from + // the xtable.hudi.target.table_version config; defaults to version 9. + private HoodieTableVersion tableVersion = HudiTargetConfig.DEFAULT_TABLE_VERSION; public HudiConversionTarget() {} @@ -131,6 +135,8 @@ public HudiConversionTarget( HudiTableManager.of(configuration), CommitState::new); this.databaseName = resolveDatabaseName(targetTable); + this.tableVersion = + HudiTargetConfig.fromProperties(targetTable.getAdditionalProperties()).getTableVersion(); } @VisibleForTesting @@ -185,6 +191,8 @@ public void init(TargetTable targetTable, Configuration configuration) { HudiTableManager.of(configuration), CommitState::new); this.databaseName = resolveDatabaseName(targetTable); + this.tableVersion = + HudiTargetConfig.fromProperties(targetTable.getAdditionalProperties()).getTableVersion(); } /** Uses the first namespace level as the Hudi database name, or the default if none is set. */ @@ -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)); } else { // make sure meta client has up-to-date view of the timeline getMetaClient().reloadActiveTimeline(); @@ -572,14 +582,15 @@ private void markInstantsAsCleaned( entry.getValue().stream() .map(HoodieCleanFileInfo::getFilePath) .collect(Collectors.toList()); - return new HoodieCleanStat( - HoodieCleaningPolicy.KEEP_LATEST_COMMITS, - partitionPath, - deletePaths, - deletePaths, - Collections.emptyList(), - earliestInstant.get().requestedTime(), - instantTime); + return HoodieCleanStat.builder() + .withPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS) + .withPartitionPath(partitionPath) + .withDeletePathPatterns(deletePaths) + .withSuccessDeleteFiles(deletePaths) + .withFailedDeleteFiles(Collections.emptyList()) + .withEarliestCommitToRetain(earliestInstant.get().requestedTime()) + .withLastCompletedCommitTimestamp(instantTime) + .build(); }) .collect(Collectors.toList()); HoodieCleanMetadata cleanMetadata = @@ -599,9 +610,13 @@ private void markInstantsAsCleaned( private void runArchiver( HoodieJavaTable table, HoodieWriteConfig config, HoodieEngineContext engineContext) { - // trigger archiver manually + // trigger archiver manually, selecting the archiver implementation that matches the table's + // timeline layout (V1 for table version 6, V2/LSM for table version 9). try { - HoodieTimelineArchiver archiver = new TimelineArchiverV1(config, table); + @SuppressWarnings({"unchecked", "rawtypes"}) + HoodieTimelineArchiver archiver = + TimelineArchivers.getInstance( + table.getMetaClient().getTimelineLayoutVersion(), config, (HoodieTable) table); archiver.archiveIfRequired(engineContext, true); } catch (IOException ex) { throw new UpdateException("Unable to archive Hudi timeline", ex); @@ -622,10 +637,11 @@ private HoodieWriteConfig getWriteConfig( Properties properties = new Properties(); properties.setProperty(HoodieMetadataConfig.AUTO_INITIALIZE.key(), "false"); return HoodieWriteConfig.newBuilder() - // Pin writes to table version 6 and disable auto-upgrade so the write client does not - // upgrade the table to version 9 (Hudi 1.x). Table version 9 support will be added in a - // follow-up PR, tracked in https://github.com/apache/incubator-xtable/issues/834. - .withWriteTableVersion(HoodieTableVersion.SIX.versionCode()) + // Write at the table's own format version (selected via xtable.hudi.target.table_version, + // default 9) and disable auto-upgrade so the write client never migrates the table to a + // different version behind our back. See + // https://github.com/apache/incubator-xtable/issues/834. + .withWriteTableVersion(metaClient.getTableConfig().getTableVersion().versionCode()) .withAutoUpgradeVersion(false) .withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(INMEMORY).build()) .withPath(metaClient.getBasePath().toString()) @@ -648,14 +664,17 @@ 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 + // in https://github.com/apache/incubator-xtable/issues/832. + .withMetadataIndexColumnStats(true) + .withMetadataIndexPartitionStats(false) .withMaxNumDeltaCommitsBeforeCompaction(maxNumDeltaCommitsBeforeCompaction) .build()) .build(); diff --git a/xtable-core/src/main/java/org/apache/xtable/hudi/HudiFileStatsExtractor.java b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiFileStatsExtractor.java index 708b336cd..c50b2eb2c 100644 --- a/xtable-core/src/main/java/org/apache/xtable/hudi/HudiFileStatsExtractor.java +++ b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiFileStatsExtractor.java @@ -54,9 +54,9 @@ import org.apache.hudi.metadata.HoodieTableMetadata; import org.apache.hudi.metadata.HoodieTableMetadataUtil; import org.apache.hudi.metadata.MetadataPartitionType; -import org.apache.hudi.stats.HoodieColumnRangeMetadata; -import org.apache.hudi.stats.ValueMetadata; -import org.apache.hudi.stats.XTableValueMetadata; +import org.apache.hudi.metadata.stats.HoodieColumnRangeMetadata; +import org.apache.hudi.metadata.stats.ValueMetadata; +import org.apache.hudi.metadata.stats.XTableValueMetadata; import org.apache.xtable.avro.AvroSchemaConverter; import org.apache.xtable.collectors.CustomCollectors; diff --git a/xtable-core/src/main/java/org/apache/xtable/hudi/HudiTableManager.java b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiTableManager.java index b6203d905..477840eac 100644 --- a/xtable-core/src/main/java/org/apache/xtable/hudi/HudiTableManager.java +++ b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiTableManager.java @@ -92,10 +92,14 @@ public Optional loadTableMetaClientIfExists(String tableD * @param table the table to initialize * @param databaseName the database to register the table under; when null/empty the table falls * back to {@link #DEFAULT_DATABASE_NAME} + * @param tableVersion the Hudi table format version to initialize the table with * @return {@link HoodieTableMetaClient} for the table that was created */ HoodieTableMetaClient initializeHudiTable( - String tableDataPath, InternalTable table, String databaseName) { + String tableDataPath, + InternalTable table, + String databaseName, + HoodieTableVersion tableVersion) { String recordKeyField = ""; if (table.getReadSchema() != null) { List recordKeys = @@ -119,10 +123,10 @@ HoodieTableMetaClient initializeHudiTable( .setCommitTimezone(HoodieTimelineTimeZone.UTC) .setHiveStylePartitioningEnable(hiveStylePartitioningEnabled) .setTableType(HoodieTableType.COPY_ON_WRITE) - // Pin new tables to table version 6 for now. Table version 9 (Hudi 1.x) support will be - // added in a follow-up PR, tracked in + // Table format version (6 = legacy 0.x layout, 9 = Hudi 1.x layout) is selected via the + // xtable.hudi.target.table_version config and defaults to 9. See // https://github.com/apache/incubator-xtable/issues/834. - .setTableVersion(HoodieTableVersion.SIX) + .setTableVersion(tableVersion) .setTableName(table.getName()) .setDatabaseName(resolvedDatabaseName) .setPayloadClass(HoodieAvroPayload.class) diff --git a/xtable-core/src/main/java/org/apache/xtable/hudi/HudiTargetConfig.java b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiTargetConfig.java new file mode 100644 index 000000000..3e2a4dc37 --- /dev/null +++ b/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}. + */ + public static final String HUDI_TABLE_VERSION = "xtable.hudi.target.table_version"; + + static final HoodieTableVersion DEFAULT_TABLE_VERSION = HoodieTableVersion.SIX; + + HoodieTableVersion tableVersion; + + public static HudiTargetConfig fromProperties(Properties properties) { + HoodieTableVersion tableVersion = DEFAULT_TABLE_VERSION; + if (properties != null) { + String configured = properties.getProperty(HUDI_TABLE_VERSION); + if (configured != null && !configured.trim().isEmpty()) { + tableVersion = HoodieTableVersion.fromVersionCode(Integer.parseInt(configured.trim())); + } + } + if (tableVersion != HoodieTableVersion.SIX && tableVersion != HoodieTableVersion.NINE) { + throw new IllegalArgumentException( + String.format( + "Unsupported Hudi target table version %s. Only table versions 6 and 9 are supported via %s.", + tableVersion.versionCode(), HUDI_TABLE_VERSION)); + } + return new HudiTargetConfig(tableVersion); + } +} diff --git a/xtable-core/src/test/java/org/apache/hudi/stats/TestXTableValueMetadata.java b/xtable-core/src/test/java/org/apache/hudi/metadata/stats/TestXTableValueMetadata.java similarity index 99% rename from xtable-core/src/test/java/org/apache/hudi/stats/TestXTableValueMetadata.java rename to xtable-core/src/test/java/org/apache/hudi/metadata/stats/TestXTableValueMetadata.java index 09dabd3b4..da585bdb5 100644 --- a/xtable-core/src/test/java/org/apache/hudi/stats/TestXTableValueMetadata.java +++ b/xtable-core/src/test/java/org/apache/hudi/metadata/stats/TestXTableValueMetadata.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.stats; +package org.apache.hudi.metadata.stats; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; diff --git a/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java b/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java index b864e07a8..a4583ed47 100644 --- a/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java +++ b/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java @@ -80,6 +80,7 @@ import org.apache.hudi.common.model.HoodieAvroPayload; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.table.HoodieTableVersion; import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.iceberg.Snapshot; @@ -101,6 +102,7 @@ import org.apache.xtable.conversion.TargetTable; import org.apache.xtable.delta.DeltaConversionSourceProvider; import org.apache.xtable.hudi.HudiConversionSourceProvider; +import org.apache.xtable.hudi.HudiTargetConfig; import org.apache.xtable.hudi.HudiTestUtil; import org.apache.xtable.iceberg.IcebergConversionSourceProvider; import org.apache.xtable.iceberg.TestIcebergDataHelper; @@ -153,7 +155,16 @@ private static Stream 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. + arguments.add(Arguments.of(sourceFormat, syncMode, isPartitioned, null)); + } else { + // Hudi is one of the targets; exercise both supported target versions. + arguments.add( + Arguments.of(sourceFormat, syncMode, isPartitioned, HoodieTableVersion.SIX)); + arguments.add( + Arguments.of(sourceFormat, syncMode, isPartitioned, HoodieTableVersion.NINE)); + } } } } @@ -225,14 +236,12 @@ private ConversionSourceProvider getConversionSourceProvider(String sourceTab @ParameterizedTest @MethodSource("generateTestParametersForFormatsSyncModesAndPartitioning") public void testVariousOperations( - String sourceTableFormat, SyncMode syncMode, boolean isPartitioned) { + String sourceTableFormat, + SyncMode syncMode, + boolean isPartitioned, + HoodieTableVersion hudiTargetVersion) { String tableName = getTableName(); List targetTableFormats = getOtherFormats(sourceTableFormat); - if (sourceTableFormat.equals(PAIMON)) { - // TODO: Hudi 1.x target is not supported for un-partitioned Paimon source. - targetTableFormats = - targetTableFormats.stream().filter(fmt -> !fmt.equals(HUDI)).collect(Collectors.toList()); - } String partitionConfig = null; if (isPartitioned) { partitionConfig = "level:VALUE"; @@ -253,7 +262,8 @@ public void testVariousOperations( table, targetTableFormats, partitionConfig, - null); + null, + hudiTargetVersion); conversionController.sync(conversionConfig, conversionSourceProvider); checkDatasetEquivalence(sourceTableFormat, table, targetTableFormats, 100); @@ -285,7 +295,8 @@ public void testVariousOperations( tableWithUpdatedSchema, targetTableFormats, partitionConfig, - null); + null, + hudiTargetVersion); List insertsAfterSchemaUpdate = tableWithUpdatedSchema.insertRows(100); tableWithUpdatedSchema.reload(); conversionController.sync(conversionConfig, conversionSourceProvider); @@ -529,48 +540,53 @@ private static Stream provideArgsForPartitionTesting() { String severityFilter = "severity = 1"; String timestampAndLevelFilter = String.format("%s and %s", timestampFilter, levelFilter); return Stream.of( - Arguments.of( buildArgsForPartition( - HUDI, Arrays.asList(ICEBERG, DELTA), "level:SIMPLE", "level:VALUE", levelFilter)), - Arguments.of( + HUDI, Arrays.asList(ICEBERG, DELTA), "level:SIMPLE", "level:VALUE", levelFilter), + buildArgsForPartition( + DELTA, Arrays.asList(ICEBERG, HUDI), null, "level:VALUE", levelFilter), buildArgsForPartition( - DELTA, Arrays.asList(ICEBERG, HUDI), null, "level:VALUE", levelFilter)), - Arguments.of( + ICEBERG, Arrays.asList(DELTA, HUDI), null, "level:VALUE", levelFilter), + // Delta is excluded here since it does not support nested partition columns. buildArgsForPartition( - ICEBERG, Arrays.asList(DELTA, HUDI), null, "level:VALUE", levelFilter)), - // TODO(hudi-1.2): re-enable the nested partition column case (HUDI -> ICEBERG partitioned - // on - // "nested_record.level"). Hudi 1.2's HoodieFileGroupReaderBasedFileFormat is the only batch - // reader and it converts the partition column into a top-level Avro field named - // "nested_record.level", which Avro rejects ("Illegal character in: nested_record.level"). - // Delta is excluded here anyway since it does not support nested partition columns. - // Arguments.of( - // buildArgsForPartition( - // HUDI, - // Arrays.asList(ICEBERG), - // "nested_record.level:SIMPLE", - // "nested_record.level:VALUE", - // nestedLevelFilter)), - Arguments.of( + HUDI, + Arrays.asList(ICEBERG), + "nested_record.level:SIMPLE", + "nested_record.level:VALUE", + nestedLevelFilter), buildArgsForPartition( HUDI, Arrays.asList(ICEBERG, DELTA), "severity:SIMPLE", "severity:VALUE", - severityFilter)), - Arguments.of( + severityFilter), buildArgsForPartition( HUDI, Arrays.asList(ICEBERG, DELTA), "timestamp_micros_nullable_field:TIMESTAMP,level:SIMPLE", "timestamp_micros_nullable_field:DAY:yyyy/MM/dd,level:VALUE", timestampAndLevelFilter, - getAdditionalHudiReadOptions()))); + getAdditionalHudiReadOptions())) + .flatMap(ITConversionController::withHudiTargetVersions); + } + + /** + * Expands a partition-test case across both Hudi target versions (6 and 9) when Hudi is one of + * the target formats; otherwise yields the single case with no version override. + */ + private static Stream withHudiTargetVersions(TableFormatPartitionDataHolder holder) { + if (holder.getTargetTableFormats().contains(HUDI)) { + return Stream.of( + Arguments.of(holder, HoodieTableVersion.SIX), + Arguments.of(holder, HoodieTableVersion.NINE)); + } + return Stream.of(Arguments.of(holder, (HoodieTableVersion) null)); } @ParameterizedTest @MethodSource("provideArgsForPartitionTesting") - public void testPartitionedData(TableFormatPartitionDataHolder tableFormatPartitionDataHolder) { + public void testPartitionedData( + TableFormatPartitionDataHolder tableFormatPartitionDataHolder, + HoodieTableVersion hudiTargetVersion) { String tableName = getTableName(); String sourceTableFormat = tableFormatPartitionDataHolder.getSourceTableFormat(); List targetTableFormats = tableFormatPartitionDataHolder.getTargetTableFormats(); @@ -597,7 +613,8 @@ public void testPartitionedData(TableFormatPartitionDataHolder tableFormatPartit table, targetTableFormats, xTablePartitionConfig, - null); + null, + hudiTargetVersion); tableToClose.insertRows(100); conversionController.sync(conversionConfig, conversionSourceProvider); // Do a second sync to force the test to read back the metadata it wrote earlier @@ -1194,6 +1211,26 @@ private static ConversionConfig getTableSyncConfig( List targetTableFormats, String partitionConfig, Duration metadataRetention) { + return getTableSyncConfig( + sourceTableFormat, + syncMode, + tableName, + table, + targetTableFormats, + partitionConfig, + metadataRetention, + null); + } + + private static ConversionConfig getTableSyncConfig( + String sourceTableFormat, + SyncMode syncMode, + String tableName, + GenericTable table, + List targetTableFormats, + String partitionConfig, + Duration metadataRetention, + HoodieTableVersion hudiTargetVersion) { Properties sourceProperties = new Properties(); if (partitionConfig != null) { sourceProperties.put(PARTITION_FIELD_SPEC_CONFIG, partitionConfig); @@ -1217,7 +1254,7 @@ private static ConversionConfig getTableSyncConfig( // set the metadata path to the data path as the default (required by Hudi) .basePath(table.getDataPath()) .metadataRetention(metadataRetention) - .additionalProperties(new TypedProperties()) + .additionalProperties(hudiTargetProperties(formatName, hudiTargetVersion)) .build()) .collect(Collectors.toList()); @@ -1227,4 +1264,18 @@ private static ConversionConfig getTableSyncConfig( .syncMode(syncMode) .build(); } + + /** + * Returns the additional properties for a target table, pinning the Hudi target table version + * when one is supplied (and the target is Hudi) so a single test can exercise both v6 and v9. + */ + private static TypedProperties hudiTargetProperties( + String formatName, HoodieTableVersion hudiTargetVersion) { + TypedProperties properties = new TypedProperties(); + if (HUDI.equals(formatName) && hudiTargetVersion != null) { + properties.setProperty( + HudiTargetConfig.HUDI_TABLE_VERSION, String.valueOf(hudiTargetVersion.versionCode())); + } + return properties; + } } diff --git a/xtable-core/src/test/java/org/apache/xtable/TestAbstractHudiTable.java b/xtable-core/src/test/java/org/apache/xtable/TestAbstractHudiTable.java index a5909a04c..242b8580c 100644 --- a/xtable-core/src/test/java/org/apache/xtable/TestAbstractHudiTable.java +++ b/xtable-core/src/test/java/org/apache/xtable/TestAbstractHudiTable.java @@ -142,6 +142,11 @@ public abstract class TestAbstractHudiTable protected KeyGenerator keyGenerator; protected Schema schema; protected List partitionFieldNames; + // Hudi table format version used when creating the test table. Defaults to version 6 (legacy 0.x + // timeline layout / column-stats index V1) to match the historical behaviour; tests covering the + // Hudi 1.x layout (column-stats index V2) can opt into version 9. See + // https://github.com/apache/incubator-xtable/issues/834. + protected HoodieTableVersion tableVersion = HoodieTableVersion.SIX; TestAbstractHudiTable(String name, Schema schema, Path tempDir, String partitionConfig) { try { @@ -441,10 +446,6 @@ protected HoodieWriteConfig generateWriteConfig(Schema schema, TypedProperties k HoodieStorageConfig.newBuilder().parquetCompressionCodec("UNCOMPRESSED").build(); HoodieArchivalConfig archivalConfig = HoodieArchivalConfig.newBuilder().archiveCommitsWith(3, 4).build(); - // Hudi 1.x MDT col-stats generation fails for array and map types, so only enable column - // stats when the schema does not contain those types. - // https://github.com/apache/incubator-xtable/issues/773 - // boolean columnStatsSupported = !schemaContainsArrayOrMap(schema); HoodieMetadataConfig metadataConfig = HoodieMetadataConfig.newBuilder() .enable(true) @@ -457,10 +458,9 @@ protected HoodieWriteConfig generateWriteConfig(Schema schema, TypedProperties k LockConfiguration.LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, "3000"); lockProperties.setProperty(LockConfiguration.LOCK_ACQUIRE_CLIENT_NUM_RETRIES_PROP_KEY, "20"); return HoodieWriteConfig.newBuilder() - // Pin writes to table version 6 and disable auto-upgrade so the write client does not - // upgrade the test table to version 9. Table version 9 support will be added in a - // follow-up PR. - .withWriteTableVersion(HoodieTableVersion.SIX.versionCode()) + // Write at the configured table version (default 6) and disable auto-upgrade so the write + // client does not migrate the test table to a different version. + .withWriteTableVersion(tableVersion.versionCode()) .withAutoUpgradeVersion(false) .withProperties(keyGenProperties) .withPath(this.basePath) @@ -629,9 +629,9 @@ protected HoodieTableMetaClient getMetaClient( .set(keyGenPropsMap) .setTableName(tableName) .setTableType(hoodieTableType) - // Pin test tables to table version 6 to match the conversion target. Table version 9 - // support will be added in a follow-up PR. - .setTableVersion(HoodieTableVersion.SIX) + // Use the configured table version (default 6) so tests can exercise both the legacy and + // Hudi 1.x layouts. + .setTableVersion(tableVersion) .setKeyGeneratorClassProp(keyGenerator.getClass().getCanonicalName()) .setPartitionFields(String.join(",", partitionFieldNames)) .setRecordKeyFields(RECORD_KEY_FIELD_NAME) diff --git a/xtable-core/src/test/java/org/apache/xtable/TestJavaHudiTable.java b/xtable-core/src/test/java/org/apache/xtable/TestJavaHudiTable.java index 499ac08f8..424b76a1f 100644 --- a/xtable-core/src/test/java/org/apache/xtable/TestJavaHudiTable.java +++ b/xtable-core/src/test/java/org/apache/xtable/TestJavaHudiTable.java @@ -39,12 +39,12 @@ import org.apache.avro.generic.GenericRecord; import org.apache.hadoop.conf.Configuration; -import org.apache.hudi.avro.HoodieAvroUtils; import org.apache.hudi.client.HoodieJavaWriteClient; import org.apache.hudi.client.WriteStatus; import org.apache.hudi.client.clustering.plan.strategy.JavaSizeBasedClusteringPlanStrategy; import org.apache.hudi.client.clustering.run.strategy.JavaSortAndSizeExecutionStrategy; import org.apache.hudi.client.common.HoodieJavaEngineContext; +import org.apache.hudi.common.avro.HoodieAvroUtils; import org.apache.hudi.common.config.TypedProperties; import org.apache.hudi.common.engine.HoodieEngineContext; import org.apache.hudi.common.model.HoodieAvroPayload; @@ -53,6 +53,7 @@ import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; import org.apache.hudi.common.util.Option; import org.apache.hudi.config.HoodieArchivalConfig; import org.apache.hudi.config.HoodieClusteringConfig; @@ -89,6 +90,16 @@ public static TestJavaHudiTable forStandardSchema( tableName, BASIC_SCHEMA, tempDir, partitionConfig, tableType, null, false); } + public static TestJavaHudiTable forStandardSchema( + String tableName, + Path tempDir, + String partitionConfig, + HoodieTableType tableType, + HoodieTableVersion tableVersion) { + return new TestJavaHudiTable( + tableName, BASIC_SCHEMA, tempDir, partitionConfig, tableType, null, false, tableVersion); + } + public static TestJavaHudiTable forStandardSchemaWithFieldIds( String tableName, Path tempDir, String partitionConfig, HoodieTableType tableType) { return new TestJavaHudiTable( @@ -170,6 +181,17 @@ public static TestJavaHudiTable withSchema( tableName, schema, tempDir, partitionConfig, tableType, null, false); } + public static TestJavaHudiTable withSchema( + String tableName, + Path tempDir, + String partitionConfig, + HoodieTableType tableType, + Schema schema, + HoodieTableVersion tableVersion) { + return new TestJavaHudiTable( + tableName, schema, tempDir, partitionConfig, tableType, null, false, tableVersion); + } + private TestJavaHudiTable( String name, Schema schema, @@ -178,7 +200,28 @@ private TestJavaHudiTable( HoodieTableType hoodieTableType, HoodieArchivalConfig archivalConfig, boolean addFieldIds) { + this( + name, + schema, + tempDir, + partitionConfig, + hoodieTableType, + archivalConfig, + addFieldIds, + HoodieTableVersion.SIX); + } + + private TestJavaHudiTable( + String name, + Schema schema, + Path tempDir, + String partitionConfig, + HoodieTableType hoodieTableType, + HoodieArchivalConfig archivalConfig, + boolean addFieldIds, + HoodieTableVersion tableVersion) { super(name, schema, tempDir, partitionConfig); + this.tableVersion = tableVersion; this.conf = new Configuration(); this.conf.set("parquet.avro.write-old-list-structure", "false"); this.addFieldIds = addFieldIds; @@ -202,6 +245,21 @@ public List> insertRecordsWithCommitAlreadyStart return inserts; } + /** + * Writes the records for an already-started commit without finalizing it. Pair with {@link + * #commitInstant} to control completion order independently of the requested (instant) time, e.g. + * to reproduce out-of-order completion on table version 9. + */ + public List bulkInsertWithoutCommit( + List> inserts, String commitInstant) { + return writeClient.bulkInsert(copyRecords(inserts), commitInstant); + } + + public void commitInstant(String commitInstant, List writeStatuses) { + writeClient.commit(commitInstant, writeStatuses); + assertNoWriteErrors(writeStatuses); + } + public List> upsertRecordsWithCommitAlreadyStarted( List> records, String commitInstant, diff --git a/xtable-core/src/test/java/org/apache/xtable/TestSparkHudiTable.java b/xtable-core/src/test/java/org/apache/xtable/TestSparkHudiTable.java index 6a6f17305..ebeef2e54 100644 --- a/xtable-core/src/test/java/org/apache/xtable/TestSparkHudiTable.java +++ b/xtable-core/src/test/java/org/apache/xtable/TestSparkHudiTable.java @@ -47,6 +47,7 @@ import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.model.WriteOperationType; import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; import org.apache.hudi.common.util.CommitUtils; import org.apache.hudi.common.util.Option; import org.apache.hudi.config.HoodieWriteConfig; @@ -115,6 +116,17 @@ public static TestSparkHudiTable forStandardSchema( tableName, BASIC_SCHEMA, tempDir, jsc, partitionConfig, tableType); } + public static TestSparkHudiTable forStandardSchema( + String tableName, + Path tempDir, + JavaSparkContext jsc, + String partitionConfig, + HoodieTableType tableType, + HoodieTableVersion tableVersion) { + return new TestSparkHudiTable( + tableName, BASIC_SCHEMA, tempDir, jsc, partitionConfig, tableType, tableVersion); + } + /** * Create a test table instance with a schema that has more fields than an instance returned by * {@link #forStandardSchema(String, Path, JavaSparkContext, String, HoodieTableType)}. @@ -153,7 +165,20 @@ private TestSparkHudiTable( JavaSparkContext jsc, String partitionConfig, HoodieTableType hoodieTableType) { + this(name, schema, tempDir, jsc, partitionConfig, hoodieTableType, HoodieTableVersion.SIX); + } + + private TestSparkHudiTable( + String name, + Schema schema, + Path tempDir, + JavaSparkContext jsc, + String partitionConfig, + HoodieTableType hoodieTableType, + HoodieTableVersion tableVersion) { super(name, schema, tempDir, partitionConfig); + // set the table version before initializing the write/meta clients, which read it + this.tableVersion = tableVersion; // initialize spark session this.jsc = jsc; this.writeClient = initSparkWriteClient(schema, typedProperties); diff --git a/xtable-core/src/test/java/org/apache/xtable/hudi/HudiTestUtil.java b/xtable-core/src/test/java/org/apache/xtable/hudi/HudiTestUtil.java index e64128e22..ad17c2a56 100644 --- a/xtable-core/src/test/java/org/apache/xtable/hudi/HudiTestUtil.java +++ b/xtable-core/src/test/java/org/apache/xtable/hudi/HudiTestUtil.java @@ -48,7 +48,7 @@ import org.apache.hudi.config.HoodieArchivalConfig; import org.apache.hudi.config.HoodieIndexConfig; import org.apache.hudi.config.HoodieWriteConfig; -import org.apache.hudi.stats.HoodieColumnRangeMetadata; +import org.apache.hudi.metadata.stats.HoodieColumnRangeMetadata; @NoArgsConstructor(access = AccessLevel.PRIVATE) public class HudiTestUtil { diff --git a/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSource.java b/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSource.java index 3046834b6..166412177 100644 --- a/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSource.java +++ b/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSource.java @@ -21,7 +21,9 @@ import static java.util.stream.Collectors.groupingBy; import static org.apache.hudi.hadoop.fs.HadoopFSUtils.getStorageConf; import static org.apache.xtable.testutil.ITTestUtils.validateTable; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertIterableEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; @@ -57,11 +59,13 @@ import org.junit.jupiter.params.provider.MethodSource; import org.apache.hudi.client.HoodieReadClient; +import org.apache.hudi.client.WriteStatus; import org.apache.hudi.common.fs.FSUtils; import org.apache.hudi.common.model.HoodieAvroPayload; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.xtable.GenericTable; @@ -230,11 +234,13 @@ void getCurrentTableTest() { @ParameterizedTest @MethodSource("testsForAllTableTypesAndPartitions") public void insertAndUpsertData( - HoodieTableType tableType, HudiTestUtil.PartitionConfig partitionConfig) { + HoodieTableType tableType, + HudiTestUtil.PartitionConfig partitionConfig, + HoodieTableVersion tableVersion) { String tableName = GenericTable.getTableName(); - try (TestJavaHudiTable table = - TestJavaHudiTable.forStandardSchema( - tableName, tempDir, partitionConfig.getHudiConfig(), tableType)) { + try (TestSparkHudiTable table = + TestSparkHudiTable.forStandardSchema( + tableName, tempDir, jsc, partitionConfig.getHudiConfig(), tableType, tableVersion)) { List> allBaseFilePaths = new ArrayList<>(); List allTableChanges = new ArrayList<>(); @@ -284,14 +290,102 @@ public void insertAndUpsertData( } } + /** + * On table version 9 (timeline layout V2) a commit becomes visible at its completion time. This + * test creates a commit whose requested (instant) time is older than a later commit but whose + * completion is newer (out-of-order completion) and verifies the incremental backlog still + * surfaces it. With the legacy requested-time selection the straggler would be skipped because + * its requested time precedes the checkpoint. + */ + @Test + public void testIncrementalSyncWithOutOfOrderCompletionOnTableVersionNine() { + String tableName = GenericTable.getTableName(); + try (TestJavaHudiTable table = + TestJavaHudiTable.forStandardSchema( + tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE, HoodieTableVersion.NINE)) { + // Baseline commit that the incremental sync resumes from. + String baseInstant = table.startCommit(); + table.insertRecordsWithCommitAlreadyStarted(table.generateRecords(20), baseInstant, true); + + // "early" is requested first (smaller instant time) but completed last; "late" is requested + // second but completed first -> completion order is the reverse of requested order. + String earlyRequestedInstant = table.startCommit(); + List earlyStatuses = + table.bulkInsertWithoutCommit(table.generateRecords(20), earlyRequestedInstant); + String lateRequestedInstant = table.startCommit(); + List lateStatuses = + table.bulkInsertWithoutCommit(table.generateRecords(20), lateRequestedInstant); + table.commitInstant(lateRequestedInstant, lateStatuses); + table.commitInstant(earlyRequestedInstant, earlyStatuses); + + HudiConversionSource hudiClient = getHudiSourceClient(CONFIGURATION, table.getBasePath(), ""); + + // Resume from the later-requested commit's instant time. Under requested-time selection the + // earlier-requested straggler (earlyRequestedInstant < lateRequestedInstant) would be + // dropped; completion-time selection must still return it. + InstantsForIncrementalSync instantsForIncrementalSync = + InstantsForIncrementalSync.builder() + .lastSyncInstant(HudiInstantUtils.parseFromInstantTime(lateRequestedInstant)) + .build(); + CommitsBacklog backlog = + hudiClient.getCommitsBacklog(instantsForIncrementalSync); + List backlogInstants = + backlog.getCommitsToProcess().stream() + .map(HoodieInstant::requestedTime) + .collect(Collectors.toList()); + + assertEquals( + Collections.singletonList(earlyRequestedInstant), + backlogInstants, + "Out-of-order completed commit must be included in the incremental backlog"); + + // The current snapshot must reflect the most-recently-completed commit (the straggler). + InternalSnapshot snapshot = hudiClient.getCurrentSnapshot(); + assertEquals( + earlyRequestedInstant, + HudiInstantUtils.convertInstantToCommit(snapshot.getTable().getLatestCommitTime())); + } + } + + /** + * Sanity check that ordinary in-order incremental sync on table version 9 returns every commit in + * completion-time order. + */ + @Test + public void testIncrementalSyncOrderingOnTableVersionNine() { + String tableName = GenericTable.getTableName(); + try (TestJavaHudiTable table = + TestJavaHudiTable.forStandardSchema( + tableName, tempDir, null, HoodieTableType.COPY_ON_WRITE, HoodieTableVersion.NINE)) { + String baseInstant = table.startCommit(); + table.insertRecordsWithCommitAlreadyStarted(table.generateRecords(20), baseInstant, true); + String second = table.startCommit(); + table.insertRecordsWithCommitAlreadyStarted(table.generateRecords(20), second, true); + String third = table.startCommit(); + table.insertRecordsWithCommitAlreadyStarted(table.generateRecords(20), third, true); + + HudiConversionSource hudiClient = getHudiSourceClient(CONFIGURATION, table.getBasePath(), ""); + CommitsBacklog backlog = + hudiClient.getCommitsBacklog( + InstantsForIncrementalSync.builder() + .lastSyncInstant(HudiInstantUtils.parseFromInstantTime(baseInstant)) + .build()); + assertIterableEquals( + Arrays.asList(second, third), + backlog.getCommitsToProcess().stream() + .map(HoodieInstant::requestedTime) + .collect(Collectors.toList())); + } + } + @Test public void testOnlyUpsertsAfterInserts() { HoodieTableType tableType = HoodieTableType.MERGE_ON_READ; HudiTestUtil.PartitionConfig partitionConfig = HudiTestUtil.PartitionConfig.of(null, null); String tableName = "test_table_" + UUID.randomUUID(); - try (TestJavaHudiTable table = - TestJavaHudiTable.forStandardSchema( - tableName, tempDir, partitionConfig.getHudiConfig(), tableType)) { + try (TestSparkHudiTable table = + TestSparkHudiTable.forStandardSchema( + tableName, tempDir, jsc, partitionConfig.getHudiConfig(), tableType)) { List> allBaseFilePaths = new ArrayList<>(); List allTableChanges = new ArrayList<>(); @@ -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 = + TestSparkHudiTable.forStandardSchema( + tableName, tempDir, jsc, partitionConfig.getHudiConfig(), tableType)) { String commitInstant1 = table.startCommit(); List> insertsForCommit1 = table.generateRecords(100); table.insertRecordsWithCommitAlreadyStarted(insertsForCommit1, commitInstant1, true); @@ -373,10 +467,11 @@ public void testForIncrementalSyncSafetyCheck() { @ParameterizedTest @MethodSource("testsForAllTableTypes") - public void testsForDropPartition(HoodieTableType tableType) { + public void testsForDropPartition(HoodieTableType tableType, HoodieTableVersion tableVersion) { String tableName = "test_table_" + UUID.randomUUID(); try (TestSparkHudiTable table = - TestSparkHudiTable.forStandardSchema(tableName, tempDir, jsc, "level:SIMPLE", tableType)) { + TestSparkHudiTable.forStandardSchema( + tableName, tempDir, jsc, "level:SIMPLE", tableType, tableVersion)) { List> allBaseFilePaths = new ArrayList<>(); List allTableChanges = new ArrayList<>(); @@ -422,10 +517,12 @@ public void testsForDropPartition(HoodieTableType tableType) { @ParameterizedTest @MethodSource("testsForAllTableTypes") - public void testMultipleInsertOverwriteOnSamePartitions(HoodieTableType tableType) { + public void testMultipleInsertOverwriteOnSamePartitions( + HoodieTableType tableType, HoodieTableVersion tableVersion) { String tableName = "test_table_" + UUID.randomUUID(); try (TestSparkHudiTable table = - TestSparkHudiTable.forStandardSchema(tableName, tempDir, jsc, "level:SIMPLE", tableType)) { + TestSparkHudiTable.forStandardSchema( + tableName, tempDir, jsc, "level:SIMPLE", tableType, tableVersion)) { List> allBaseFilePaths = new ArrayList<>(); List allTableChanges = new ArrayList<>(); @@ -470,10 +567,12 @@ public void testMultipleInsertOverwriteOnSamePartitions(HoodieTableType tableTyp @ParameterizedTest @MethodSource("testsForAllTableTypes") - public void testsForDeleteAllRecordsInPartition(HoodieTableType tableType) { + public void testsForDeleteAllRecordsInPartition( + HoodieTableType tableType, HoodieTableVersion tableVersion) { String tableName = "test_table_" + UUID.randomUUID(); try (TestSparkHudiTable table = - TestSparkHudiTable.forStandardSchema(tableName, tempDir, jsc, "level:SIMPLE", tableType)) { + TestSparkHudiTable.forStandardSchema( + tableName, tempDir, jsc, "level:SIMPLE", tableType, tableVersion)) { List> allBaseFilePaths = new ArrayList<>(); List allTableChanges = new ArrayList<>(); @@ -523,11 +622,13 @@ public void testsForDeleteAllRecordsInPartition(HoodieTableType tableType) { @ParameterizedTest @MethodSource("testsForAllTableTypesAndPartitions") public void testsForClustering( - HoodieTableType tableType, HudiTestUtil.PartitionConfig partitionConfig) { + HoodieTableType tableType, + HudiTestUtil.PartitionConfig partitionConfig, + HoodieTableVersion tableVersion) { String tableName = "test_table_" + UUID.randomUUID(); - try (TestJavaHudiTable table = - TestJavaHudiTable.forStandardSchema( - tableName, tempDir, partitionConfig.getHudiConfig(), tableType)) { + try (TestSparkHudiTable table = + TestSparkHudiTable.forStandardSchema( + tableName, tempDir, jsc, partitionConfig.getHudiConfig(), tableType, tableVersion)) { List> allBaseFilePaths = new ArrayList<>(); List allTableChanges = new ArrayList<>(); @@ -590,11 +691,13 @@ public void testsForClustering( @ParameterizedTest @MethodSource("testsForAllTableTypesAndPartitions") public void testsForSavepointRestore( - HoodieTableType tableType, HudiTestUtil.PartitionConfig partitionConfig) { + HoodieTableType tableType, + HudiTestUtil.PartitionConfig partitionConfig, + HoodieTableVersion tableVersion) { String tableName = "test_table_" + UUID.randomUUID(); - try (TestJavaHudiTable table = - TestJavaHudiTable.forStandardSchema( - tableName, tempDir, partitionConfig.getHudiConfig(), tableType)) { + try (TestSparkHudiTable table = + TestSparkHudiTable.forStandardSchema( + tableName, tempDir, jsc, partitionConfig.getHudiConfig(), tableType, tableVersion)) { List> allBaseFilePaths = new ArrayList<>(); List allTableChanges = new ArrayList<>(); @@ -657,11 +760,13 @@ public void testsForSavepointRestore( @ParameterizedTest @MethodSource("testsForAllTableTypesAndPartitions") public void testsForRollbacks( - HoodieTableType tableType, HudiTestUtil.PartitionConfig partitionConfig) { + HoodieTableType tableType, + HudiTestUtil.PartitionConfig partitionConfig, + HoodieTableVersion tableVersion) { String tableName = "test_table_" + UUID.randomUUID(); - try (TestJavaHudiTable table = - TestJavaHudiTable.forStandardSchema( - tableName, tempDir, partitionConfig.getHudiConfig(), tableType)) { + try (TestSparkHudiTable table = + TestSparkHudiTable.forStandardSchema( + tableName, tempDir, jsc, partitionConfig.getHudiConfig(), tableType, tableVersion)) { String commitInstant1 = table.startCommit(); List> insertsForCommit1 = table.generateRecords(50); @@ -718,8 +823,13 @@ public void testsForRollbacks( } private static Stream testsForAllTableTypes() { - return Stream.of( - Arguments.of(HoodieTableType.COPY_ON_WRITE), Arguments.of(HoodieTableType.MERGE_ON_READ)); + List tableTypes = + Arrays.asList(HoodieTableType.COPY_ON_WRITE, HoodieTableType.MERGE_ON_READ); + List tableVersions = + Arrays.asList(HoodieTableVersion.SIX, HoodieTableVersion.NINE); + return tableTypes.stream() + .flatMap( + tableType -> tableVersions.stream().map(version -> Arguments.of(tableType, version))); } private static Stream testsForAllTableTypesAndPartitions() { @@ -730,10 +840,17 @@ private static Stream testsForAllTableTypesAndPartitions() { Arrays.asList(unPartitionedConfig, partitionedConfig); List tableTypes = Arrays.asList(HoodieTableType.COPY_ON_WRITE, HoodieTableType.MERGE_ON_READ); + List tableVersions = + Arrays.asList(HoodieTableVersion.SIX, HoodieTableVersion.NINE); return tableTypes.stream() .flatMap( - tableType -> partitionConfigs.stream().map(config -> Arguments.of(tableType, config))); + tableType -> + partitionConfigs.stream() + .flatMap( + config -> + tableVersions.stream() + .map(version -> Arguments.of(tableType, config, version)))); } private HudiConversionSource getHudiSourceClient( diff --git a/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionTarget.java b/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionTarget.java index b0f2b87a3..4adadc8d5 100644 --- a/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionTarget.java +++ b/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionTarget.java @@ -40,6 +40,7 @@ import java.util.TimeZone; import java.util.UUID; import java.util.stream.Collectors; +import java.util.stream.Stream; import lombok.SneakyThrows; @@ -51,7 +52,9 @@ import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.api.parallel.Execution; import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.MethodSource; import org.apache.hudi.avro.model.HoodieMetadataColumnStats; import org.apache.hudi.avro.model.StringWrapper; @@ -66,6 +69,7 @@ import org.apache.hudi.common.model.HoodieTimelineTimeZone; import org.apache.hudi.common.model.WriteOperationType; import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; import org.apache.hudi.common.table.TableSchemaResolver; import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator; @@ -151,8 +155,8 @@ public static void setupOnce() { } @ParameterizedTest - @ValueSource(booleans = {true, false}) - void syncForExistingTable(boolean partitioned) { + @MethodSource("partitionedAndVersion") + void syncForExistingTable(boolean partitioned, HoodieTableVersion tableVersion) { String partitionPath = partitioned ? "partition_path" : ""; String commitTime = "20231003013807542"; String existingFileName1 = "existing_file_1.parquet"; @@ -211,7 +215,7 @@ void syncForExistingTable(boolean partitioned) { .fileRemoved(fileToRemove) .build(); // perform sync - HudiConversionTarget targetClient = getTargetClient(); + HudiConversionTarget targetClient = getTargetClient(tableVersion); InternalTable initialState = getState(Instant.now(), partitioned); targetClient.beginSync(initialState); targetClient.syncFilesForDiff(internalFilesDiff); @@ -226,24 +230,22 @@ void syncForExistingTable(boolean partitioned) { HoodieTableMetaClient.builder().setConf(CONFIGURATION).setBasePath(tableBasePath).build(); assertFileGroupCorrectness( metaClient, partitionPath, Collections.singletonList(Pair.of(fileName, filePath))); - if (!partitioned) { - try (HoodieBackedTableMetadata hoodieBackedTableMetadata = - new HoodieBackedTableMetadata( - CONTEXT, - metaClient.getStorage(), - writeConfig.getMetadataConfig(), - tableBasePath, - true)) { - assertColStats(hoodieBackedTableMetadata, partitionPath, fileName); - } + try (HoodieBackedTableMetadata hoodieBackedTableMetadata = + new HoodieBackedTableMetadata( + CONTEXT, + metaClient.getStorage(), + writeConfig.getMetadataConfig(), + tableBasePath, + true)) { + assertColStats(hoodieBackedTableMetadata, partitionPath, fileName); } // include meta fields since the table was created with meta fields enabled assertSchema(metaClient, true); } @ParameterizedTest - @ValueSource(booleans = {true, false}) - void syncForNewTable(boolean partitioned) { + @MethodSource("partitionedAndVersion") + void syncForNewTable(boolean partitioned, HoodieTableVersion tableVersion) { String partitionPath = partitioned ? "partition_path" : ""; String fileName = "file_1.parquet"; String filePath = getFilePath(partitionPath, fileName); @@ -260,7 +262,7 @@ void syncForNewTable(boolean partitioned) { .build()); // sync snapshot and metadata InternalTable initialState = getState(Instant.now(), partitioned); - HudiConversionTarget targetClient = getTargetClient(); + HudiConversionTarget targetClient = getTargetClient(tableVersion); targetClient.beginSync(initialState); targetClient.syncFilesForSnapshot(snapshot); TableSyncMetadata latestState = @@ -274,23 +276,24 @@ void syncForNewTable(boolean partitioned) { HoodieTableMetaClient.builder().setConf(CONFIGURATION).setBasePath(tableBasePath).build(); assertFileGroupCorrectness( metaClient, partitionPath, Collections.singletonList(Pair.of(fileName, filePath))); - if (!partitioned) { - try (HoodieBackedTableMetadata hoodieBackedTableMetadata = - new HoodieBackedTableMetadata( - CONTEXT, - metaClient.getStorage(), - getHoodieWriteConfig(metaClient).getMetadataConfig(), - tableBasePath, - true)) { - assertColStats(hoodieBackedTableMetadata, partitionPath, fileName); - } + try (HoodieBackedTableMetadata hoodieBackedTableMetadata = + new HoodieBackedTableMetadata( + CONTEXT, + metaClient.getStorage(), + getHoodieWriteConfig(metaClient).getMetadataConfig(), + tableBasePath, + true)) { + assertColStats(hoodieBackedTableMetadata, partitionPath, fileName); } assertSchema(metaClient, false); } @ParameterizedTest - @ValueSource(booleans = {true, false}) - void archiveTimelineAndCleanMetadataTableAfterMultipleCommits(boolean partitioned) { + @EnumSource( + value = HoodieTableVersion.class, + names = {"SIX", "NINE"}) + void archiveTimelineAndCleanMetadataTableAfterMultipleCommits(HoodieTableVersion tableVersion) { + boolean partitioned = true; String partitionPath = partitioned ? "partition_path" : ""; String fileName0 = "file_0.parquet"; String filePath0 = getFilePath(partitionPath, fileName0); @@ -313,7 +316,7 @@ void archiveTimelineAndCleanMetadataTableAfterMultipleCommits(boolean partitione .build()); // sync snapshot and metadata InternalTable initialState = getState(Instant.now().minus(24, ChronoUnit.HOURS), partitioned); - HudiConversionTarget targetClient = getTargetClient(); + HudiConversionTarget targetClient = getTargetClient(tableVersion); targetClient.beginSync(initialState); targetClient.syncFilesForSnapshot(snapshot); TableSyncMetadata latestState = @@ -328,16 +331,14 @@ void archiveTimelineAndCleanMetadataTableAfterMultipleCommits(boolean partitione Pair file0Pair = Pair.of(fileName0, filePath0); assertFileGroupCorrectness( metaClient, partitionPath, Arrays.asList(file0Pair, Pair.of(fileName1, filePath1))); - if (!partitioned) { - try (HoodieBackedTableMetadata hoodieBackedTableMetadata = - new HoodieBackedTableMetadata( - CONTEXT, - metaClient.getStorage(), - getHoodieWriteConfig(metaClient).getMetadataConfig(), - tableBasePath, - true)) { - assertColStats(hoodieBackedTableMetadata, partitionPath, fileName1); - } + try (HoodieBackedTableMetadata hoodieBackedTableMetadata = + new HoodieBackedTableMetadata( + CONTEXT, + metaClient.getStorage(), + getHoodieWriteConfig(metaClient).getMetadataConfig(), + tableBasePath, + true)) { + assertColStats(hoodieBackedTableMetadata, partitionPath, fileName1); } // create a new commit that removes fileName1 and adds fileName2 @@ -353,19 +354,17 @@ void archiveTimelineAndCleanMetadataTableAfterMultipleCommits(boolean partitione assertFileGroupCorrectness( metaClient, partitionPath, Arrays.asList(file0Pair, Pair.of(fileName2, filePath2))); - if (!partitioned) { - try (HoodieBackedTableMetadata hoodieBackedTableMetadata = - new HoodieBackedTableMetadata( - CONTEXT, - metaClient.getStorage(), - getHoodieWriteConfig(metaClient).getMetadataConfig(), - tableBasePath, - true)) { - // the metadata for fileName1 should still be present until the cleaner kicks in - assertColStats(hoodieBackedTableMetadata, partitionPath, fileName1); - // new file stats should be present - assertColStats(hoodieBackedTableMetadata, partitionPath, fileName2); - } + try (HoodieBackedTableMetadata hoodieBackedTableMetadata = + new HoodieBackedTableMetadata( + CONTEXT, + metaClient.getStorage(), + getHoodieWriteConfig(metaClient).getMetadataConfig(), + tableBasePath, + true)) { + // the metadata for fileName1 should still be present until the cleaner kicks in + assertColStats(hoodieBackedTableMetadata, partitionPath, fileName1); + // new file stats should be present + assertColStats(hoodieBackedTableMetadata, partitionPath, fileName2); } // create a new commit that removes fileName2 and adds fileName3 @@ -413,18 +412,16 @@ void archiveTimelineAndCleanMetadataTableAfterMultipleCommits(boolean partitione Pair.of(fileName4, filePath4), Pair.of(fileName5, filePath5))); // col stats should be cleaned up for fileName1 but present for fileName2 and fileName3 - if (!partitioned) { - try (HoodieBackedTableMetadata hoodieBackedTableMetadata = - new HoodieBackedTableMetadata( - CONTEXT, - metaClient.getStorage(), - getHoodieWriteConfig(metaClient).getMetadataConfig(), - tableBasePath, - true)) { - // assertEmptyColStats(hoodieBackedTableMetadata, partitionPath, fileName1); - assertColStats(hoodieBackedTableMetadata, partitionPath, fileName3); - assertColStats(hoodieBackedTableMetadata, partitionPath, fileName4); - } + try (HoodieBackedTableMetadata hoodieBackedTableMetadata = + new HoodieBackedTableMetadata( + CONTEXT, + metaClient.getStorage(), + getHoodieWriteConfig(metaClient).getMetadataConfig(), + tableBasePath, + true)) { + // assertEmptyColStats(hoodieBackedTableMetadata, partitionPath, fileName1); + assertColStats(hoodieBackedTableMetadata, partitionPath, fileName3); + assertColStats(hoodieBackedTableMetadata, partitionPath, fileName4); } // the first commit to the timeline should be archived assertEquals( @@ -432,8 +429,9 @@ void archiveTimelineAndCleanMetadataTableAfterMultipleCommits(boolean partitione } @ParameterizedTest - @ValueSource(booleans = {true, false}) - void testSourceTargetMappingWithSnapshotAndIncrementalSync(boolean partitioned) { + @MethodSource("partitionedAndVersion") + void testSourceTargetMappingWithSnapshotAndIncrementalSync( + boolean partitioned, HoodieTableVersion tableVersion) { String partitionPath = partitioned ? "partition_path" : ""; // Step 1: Initialize Test Files for Initial Snapshot String fileName0 = "file_0.parquet"; @@ -456,7 +454,7 @@ void testSourceTargetMappingWithSnapshotAndIncrementalSync(boolean partitioned) // Step 2: Sync Initial Snapshot InternalTable initialState = getState(Instant.now().minus(24, ChronoUnit.HOURS), partitioned); - HudiConversionTarget targetClient = getTargetClient(); + HudiConversionTarget targetClient = getTargetClient(tableVersion); targetClient.beginSync(initialState); targetClient.syncFilesForSnapshot(initialSnapshot); TableSyncMetadata latestState = @@ -522,8 +520,9 @@ void testSourceTargetMappingWithSnapshotAndIncrementalSync(boolean partitioned) } @ParameterizedTest - @ValueSource(booleans = {true, false}) - void testGetTargetCommitIdentifierWithNullSourceIdentifier(boolean partitioned) { + @MethodSource("partitionedAndVersion") + void testGetTargetCommitIdentifierWithNullSourceIdentifier( + boolean partitioned, HoodieTableVersion tableVersion) { String partitionPath = partitioned ? "partition_path" : ""; // Initialize Test Files and Snapshot String fileName0 = "file_0.parquet"; @@ -544,7 +543,7 @@ void testGetTargetCommitIdentifierWithNullSourceIdentifier(boolean partitioned) .build())) .build()); InternalTable internalTable = getState(Instant.now().minus(24, ChronoUnit.HOURS), partitioned); - HudiConversionTarget targetClient = getTargetClient(); + HudiConversionTarget targetClient = getTargetClient(tableVersion); targetClient.beginSync(internalTable); targetClient.syncFilesForSnapshot(initialSnapshot); @@ -785,14 +784,26 @@ private InternalTable getState(Instant latestCommitTime, boolean partitioned) { return builder.build(); } - private HudiConversionTarget getTargetClient() { + /** Cross-product of the partitioned flag with the supported Hudi table versions (6 and 9). */ + private static Stream partitionedAndVersion() { + return Stream.of( + Arguments.of(true, HoodieTableVersion.SIX), + Arguments.of(false, HoodieTableVersion.SIX), + Arguments.of(true, HoodieTableVersion.NINE), + Arguments.of(false, HoodieTableVersion.NINE)); + } + + private HudiConversionTarget getTargetClient(HoodieTableVersion tableVersion) { + TypedProperties properties = new TypedProperties(); + properties.setProperty( + HudiTargetConfig.HUDI_TABLE_VERSION, String.valueOf(tableVersion.versionCode())); return new HudiConversionTarget( TargetTable.builder() .basePath(tableBasePath) .formatName(TableFormat.HUDI) .name("test_table") .metadataRetention(Duration.of(4, ChronoUnit.HOURS)) - .additionalProperties(new TypedProperties()) + .additionalProperties(properties) .build(), (Configuration) CONFIGURATION.unwrapCopy(), 3); diff --git a/xtable-core/src/test/java/org/apache/xtable/hudi/TestBaseFileUpdatesExtractor.java b/xtable-core/src/test/java/org/apache/xtable/hudi/TestBaseFileUpdatesExtractor.java index 186ef1e01..5a36c3791 100644 --- a/xtable-core/src/test/java/org/apache/xtable/hudi/TestBaseFileUpdatesExtractor.java +++ b/xtable-core/src/test/java/org/apache/xtable/hudi/TestBaseFileUpdatesExtractor.java @@ -19,7 +19,7 @@ package org.apache.xtable.hudi; import static org.apache.hudi.hadoop.fs.HadoopFSUtils.getStorageConf; -import static org.apache.hudi.stats.XTableValueMetadata.getValueMetadata; +import static org.apache.hudi.metadata.stats.XTableValueMetadata.getValueMetadata; import static org.apache.xtable.hudi.HudiTestUtil.createWriteStatus; import static org.apache.xtable.hudi.HudiTestUtil.getHoodieWriteConfig; import static org.apache.xtable.hudi.HudiTestUtil.initTableAndGetMetaClient; @@ -60,9 +60,9 @@ import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.hadoop.fs.CachingPath; import org.apache.hudi.metadata.HoodieIndexVersion; -import org.apache.hudi.stats.HoodieColumnRangeMetadata; -import org.apache.hudi.stats.ValueMetadata; -import org.apache.hudi.stats.ValueType; +import org.apache.hudi.metadata.stats.HoodieColumnRangeMetadata; +import org.apache.hudi.metadata.stats.ValueMetadata; +import org.apache.hudi.metadata.stats.ValueType; import org.apache.xtable.model.schema.InternalField; import org.apache.xtable.model.schema.InternalPartitionField; diff --git a/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiConversionTarget.java b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiConversionTarget.java index bc9637e8a..b190d8513 100644 --- a/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiConversionTarget.java +++ b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiConversionTarget.java @@ -276,7 +276,8 @@ private Pair initMocksF when(mockMetaClient.getTableConfig()).thenReturn(mockTableConfig); when(mockTableConfig.getRecordKeyFields()) .thenReturn(Option.of(new String[] {"record_key_field"})); - when(mockHudiTableManager.initializeHudiTable(BASE_PATH, TABLE, null)) + when(mockHudiTableManager.initializeHudiTable( + BASE_PATH, TABLE, null, HudiTargetConfig.DEFAULT_TABLE_VERSION)) .thenReturn(mockMetaClient); HudiConversionTarget.CommitState mockCommitState = mock(HudiConversionTarget.CommitState.class); when(mockCommitStateCreator.create( diff --git a/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiFileStatsExtractor.java b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiFileStatsExtractor.java index d71035798..45835e906 100644 --- a/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiFileStatsExtractor.java +++ b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiFileStatsExtractor.java @@ -61,6 +61,8 @@ import org.jetbrains.annotations.NotNull; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; import org.apache.hudi.avro.model.HoodieMetadataColumnStats; import org.apache.hudi.client.common.HoodieJavaEngineContext; @@ -142,14 +144,27 @@ public class TestHudiFileStatsExtractor { decimalField)) .build(); - @Test - void columnStatsWithMetadataTable(@TempDir Path tempDir) throws Exception { + @ParameterizedTest + @EnumSource( + value = HoodieTableVersion.class, + names = {"SIX", "NINE"}) + void columnStatsWithMetadataTable(HoodieTableVersion tableVersion, @TempDir Path tempDir) + throws Exception { + // Column-stats index V1 (table version 6) excludes DECIMAL/FIXED columns (HUDI-8585), so the + // decimal_field gets no stats and 8 columns are indexed. Index V2 (table version 9) supports + // these types, so the decimal_field is present and 9 columns are indexed. See #834. + boolean includeDecimal = tableVersion == HoodieTableVersion.NINE; String tableName = GenericTable.getTableName(); String basePath; HoodieTableMetaClient metaClient; try (TestJavaHudiTable table = TestJavaHudiTable.withSchema( - tableName, tempDir, "long_field:SIMPLE", HoodieTableType.COPY_ON_WRITE, AVRO_SCHEMA)) { + tableName, + tempDir, + "long_field:SIMPLE", + HoodieTableType.COPY_ON_WRITE, + AVRO_SCHEMA, + tableVersion)) { List> records = getRecords().stream().map(this::buildRecord).collect(Collectors.toList()); table.insertRecords(true, records); @@ -182,7 +197,7 @@ void columnStatsWithMetadataTable(@TempDir Path tempDir) throws Exception { fileStatsExtractor .addStatsToFiles(tableMetadata, Stream.of(inputFile), schema) .collect(Collectors.toList()); - validateOutput(output, false); + validateOutput(output, includeDecimal); } @Test diff --git a/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiTableManager.java b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiTableManager.java index ece8f81cd..fa3a2c9bd 100644 --- a/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiTableManager.java +++ b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiTableManager.java @@ -38,6 +38,7 @@ import org.junit.jupiter.params.provider.MethodSource; import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; import org.apache.hudi.storage.StorageConfiguration; import org.apache.xtable.model.InternalTable; @@ -96,7 +97,7 @@ void validateTableInitializedCorrectly( .layoutStrategy(dataLayoutStrategy) .build(); - tableManager.initializeHudiTable(tableBasePath, table, null); + tableManager.initializeHudiTable(tableBasePath, table, null, HoodieTableVersion.NINE); HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder() @@ -140,7 +141,7 @@ void initializeHudiTableUsesProvidedDatabaseName() { .layoutStrategy(DataLayoutStrategy.FLAT) .build(); - tableManager.initializeHudiTable(tableBasePath, table, "my_namespace"); + tableManager.initializeHudiTable(tableBasePath, table, "my_namespace", HoodieTableVersion.NINE); HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder() diff --git a/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiTargetConfig.java b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiTargetConfig.java new file mode 100644 index 000000000..6cd10150c --- /dev/null +++ b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiTargetConfig.java @@ -0,0 +1,59 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.Properties; + +import org.junit.jupiter.api.Test; + +import org.apache.hudi.common.table.HoodieTableVersion; + +public class TestHudiTargetConfig { + + @Test + void defaultsToTableVersionSix() { + assertEquals( + HoodieTableVersion.SIX, + HudiTargetConfig.fromProperties(new Properties()).getTableVersion()); + assertEquals(HoodieTableVersion.SIX, HudiTargetConfig.fromProperties(null).getTableVersion()); + } + + @Test + void honoursConfiguredVersion() { + Properties sixProps = new Properties(); + sixProps.setProperty(HudiTargetConfig.HUDI_TABLE_VERSION, "6"); + assertEquals( + HoodieTableVersion.SIX, HudiTargetConfig.fromProperties(sixProps).getTableVersion()); + + Properties nineProps = new Properties(); + nineProps.setProperty(HudiTargetConfig.HUDI_TABLE_VERSION, "9"); + assertEquals( + HoodieTableVersion.NINE, HudiTargetConfig.fromProperties(nineProps).getTableVersion()); + } + + @Test + void rejectsUnsupportedVersion() { + Properties props = new Properties(); + props.setProperty(HudiTargetConfig.HUDI_TABLE_VERSION, "8"); + assertThrows(IllegalArgumentException.class, () -> HudiTargetConfig.fromProperties(props)); + } +} diff --git a/xtable-service/src/test/java/org/apache/xtable/service/ITConversionService.java b/xtable-service/src/test/java/org/apache/xtable/service/ITConversionService.java index 90b16b84b..9bdb140a4 100644 --- a/xtable-service/src/test/java/org/apache/xtable/service/ITConversionService.java +++ b/xtable-service/src/test/java/org/apache/xtable/service/ITConversionService.java @@ -157,12 +157,6 @@ public static void teardown() { public void testVariousOperations(String sourceTableFormat, boolean isPartitioned) { String tableName = getTableName(); List targetTableFormats = getOtherFormats(sourceTableFormat); - if (sourceTableFormat.equals(PAIMON)) { - // TODO: Hudi 1.x target is not supported for un-partitioned Paimon source. - // https://github.com/apache/incubator-xtable/issues/777 - targetTableFormats = - targetTableFormats.stream().filter(fmt -> !fmt.equals(HUDI)).collect(Collectors.toList()); - } String partitionConfig = isPartitioned ? "level:VALUE" : null; try (GenericTable table =