From a66698274c9126acbc345f3aee399f479e96401b Mon Sep 17 00:00:00 2001 From: Vinish Reddy Date: Mon, 29 Jun 2026 21:07:41 -0700 Subject: [PATCH 1/8] [834] Add Hudi table version 9 (Hudi 1.x) support to the Hudi target Adds an `xtable.hudi.target.table_version` config (values 6 or 9, default 9) so the Hudi target can write the Hudi 1.x table format (timeline layout V2, column-stats index V2) instead of being pinned to version 6. Target changes: - HudiTargetConfig parses the new config from the target's additional properties; HudiTableManager.initializeHudiTable and the write config now honour the selected version (write version derives from the table itself). - Enable the column-stats index for all tables and disable the partition-stats index independently (hoodie.metadata.index.partition.stats.enable, added in apache/hudi#19111) so column stats work for partitioned external-file tables. - Select the timeline archiver by layout version (TimelineArchivers.getInstance) so version 9 uses the V2/LSM archiver; switch to the HoodieCleanStat builder. Source changes: - HudiConversionSource selects and orders instants by completion time on version 9 (timeline layout V2) and by requested time on version 6, so a commit that completes out of order relative to its requested time is no longer skipped during incremental sync. Tests: - Bump hudi.version to 1.3.0-SNAPSHOT to pick up apache/hudi#19111. - Parameterize TestHudiFileStatsExtractor over versions 6 (8 columns, decimal excluded) and 9 (9 columns, decimal present). - Add TestHudiTargetConfig and an out-of-order-completion incremental sync test on a version 9 source table. --- pom.xml | 2 +- .../xtable/hudi/HudiConversionSource.java | 130 ++++++++++++++---- .../xtable/hudi/HudiConversionTarget.java | 67 +++++---- .../apache/xtable/hudi/HudiTableManager.java | 12 +- .../apache/xtable/hudi/HudiTargetConfig.java | 57 ++++++++ .../apache/xtable/TestAbstractHudiTable.java | 18 ++- .../org/apache/xtable/TestJavaHudiTable.java | 58 ++++++++ .../xtable/hudi/ITHudiConversionSource.java | 92 +++++++++++++ .../xtable/hudi/TestHudiConversionTarget.java | 2 +- .../hudi/TestHudiFileStatsExtractor.java | 23 +++- .../xtable/hudi/TestHudiTableManager.java | 5 +- .../xtable/hudi/TestHudiTargetConfig.java | 59 ++++++++ 12 files changed, 458 insertions(+), 67 deletions(-) create mode 100644 xtable-core/src/main/java/org/apache/xtable/hudi/HudiTargetConfig.java create mode 100644 xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiTargetConfig.java 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/xtable/hudi/HudiConversionSource.java b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiConversionSource.java index ecf3877e8..715bb6e74 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,81 @@ 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 orders by completion time. + */ + private List orderByCompletionTimeAndDedup( + List list1, List list2) { + Map dedupedByRequestedTime = new LinkedHashMap<>(); + Stream.concat(list1.stream(), list2.stream()) + .forEach( + hoodieInstant -> + dedupedByRequestedTime.putIfAbsent(hoodieInstant.requestedTime(), hoodieInstant)); + return dedupedByRequestedTime.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/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..3d3131ea2 --- /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.NINE; + + 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/xtable/TestAbstractHudiTable.java b/xtable-core/src/test/java/org/apache/xtable/TestAbstractHudiTable.java index a5909a04c..7a4ef3794 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 { @@ -457,10 +462,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 +633,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..8e8d05c0c 100644 --- a/xtable-core/src/test/java/org/apache/xtable/TestJavaHudiTable.java +++ b/xtable-core/src/test/java/org/apache/xtable/TestJavaHudiTable.java @@ -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/hudi/ITHudiConversionSource.java b/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSource.java index 3046834b6..979a4a636 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; @@ -284,6 +288,94 @@ 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; 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..a2a24a812 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,7 @@ 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, HoodieTableVersion.NINE)) .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..93a2f4b8b --- /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 defaultsToTableVersionNine() { + assertEquals( + HoodieTableVersion.NINE, + HudiTargetConfig.fromProperties(new Properties()).getTableVersion()); + assertEquals(HoodieTableVersion.NINE, 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)); + } +} From d37a6a250956879a328002941674f0aa12601abd Mon Sep 17 00:00:00 2001 From: Vinish Reddy Date: Tue, 30 Jun 2026 11:04:41 -0700 Subject: [PATCH 2/8] [834] Assert column stats for partitioned tables in ITHudiConversionTarget Column stats are now generated for partitioned Hudi targets (column-stats index enabled, partition-stats index disabled independently), so drop the `if (!partitioned)` guards that previously skipped the column-stats assertions for partitioned tables. The partitioned/non-partitioned parameterization is unchanged; the col-stats checks now run for both. --- .../xtable/hudi/ITHudiConversionTarget.java | 100 ++++++++---------- 1 file changed, 45 insertions(+), 55 deletions(-) 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..f75d4dd79 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 @@ -226,16 +226,14 @@ 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); @@ -274,16 +272,14 @@ 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); } @@ -328,16 +324,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 +347,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 +405,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( From ef940302ee0df90ee73f67d7673bdf73f722d7cf Mon Sep 17 00:00:00 2001 From: Vinish Reddy Date: Tue, 30 Jun 2026 12:00:24 -0700 Subject: [PATCH 3/8] [834] Drop stale array/map column-stats comment in TestAbstractHudiTable Source test tables already enable the column-stats index unconditionally and the suite passes with array/map schemas, so remove the leftover commented-out schemaContainsArrayOrMap guard and its stale #773 note. --- .../test/java/org/apache/xtable/TestAbstractHudiTable.java | 4 ---- 1 file changed, 4 deletions(-) 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 7a4ef3794..242b8580c 100644 --- a/xtable-core/src/test/java/org/apache/xtable/TestAbstractHudiTable.java +++ b/xtable-core/src/test/java/org/apache/xtable/TestAbstractHudiTable.java @@ -446,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) From 2ce1434c0e5d9da69c4923be7f07a9654cb59e4c Mon Sep 17 00:00:00 2001 From: Vinish Reddy Date: Tue, 30 Jun 2026 13:44:53 -0700 Subject: [PATCH 4/8] [834] Re-enable Paimon and nested-partition conversions to Hudi at table version 9 Re-enables the three conversion cases #772 disabled for the Hudi 1.x target: - Un-partitioned Paimon -> Hudi (ITConversionController and ITConversionService). BaseFileUpdatesExtractor now emits Hudi's external file-group-prefix format (Hudi PR #17788) for bucketed files instead of folding the "bucket-N" directory into the partition path: the file is registered under its true partition (empty for un-partitioned) with fileId "bucket-N/" and the 3-arg marker "__fg%3Dbucket-N_hudiext". Applied consistently across the snapshot path, the diff path, and file-id derivation; non-bucketed sources are unaffected. - HUDI (partitioned on the nested column "nested_record.level") -> ICEBERG. This depends on the Hudi reader fix in apache/hudi#19123, so it will only pass once that fix is available in the Hudi snapshot the build resolves. --- .../xtable/hudi/BaseFileUpdatesExtractor.java | 67 +++++++++++++++++-- .../apache/xtable/ITConversionController.java | 26 +++---- .../xtable/service/ITConversionService.java | 6 -- 3 files changed, 68 insertions(+), 31 deletions(-) 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..06c999a2e 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 @@ -75,6 +75,12 @@ 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 +240,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 +256,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 +314,30 @@ 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 +392,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/test/java/org/apache/xtable/ITConversionController.java b/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java index b864e07a8..6b0a1b210 100644 --- a/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java +++ b/xtable-core/src/test/java/org/apache/xtable/ITConversionController.java @@ -228,11 +228,6 @@ public void testVariousOperations( String sourceTableFormat, SyncMode syncMode, 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. - targetTableFormats = - targetTableFormats.stream().filter(fmt -> !fmt.equals(HUDI)).collect(Collectors.toList()); - } String partitionConfig = null; if (isPartitioned) { partitionConfig = "level:VALUE"; @@ -538,19 +533,14 @@ private static Stream provideArgsForPartitionTesting() { Arguments.of( 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)), + // Delta is excluded here 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( buildArgsForPartition( HUDI, 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 = From 979c00e021ec9bffae4d7bd8ab933c2c0821ea65 Mon Sep 17 00:00:00 2001 From: Vinish Reddy Date: Wed, 1 Jul 2026 16:28:22 -0700 Subject: [PATCH 5/8] Keep savepoint instants in the version 9 incremental backlog A savepoint instant reuses the requested time of the commit it pins, so orderByCompletionTimeAndDedup (the table version 9 / completion-time ordering path) dropped it when deduping the merged commit lists by requested time alone: putIfAbsent kept the data commit and silently discarded the savepoint. The version 6 path (mergeAndDedupLists) dedups by full instant equality, which includes the action, so it never had this problem. Include the action in the dedup key. The intended dedup (the same instant appearing in both the pending list and the newly-completed list) still collapses, but distinct actions sharing a requested time survive, and the version 9 backlog matches version 6: commit, savepoint (no-op), restore, commit. --- .../apache/xtable/hudi/HudiConversionSource.java | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) 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 715bb6e74..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 @@ -303,16 +303,21 @@ private CommitsPair getCompletedAndPendingCommitsAfterCompletionTime( } /** - * Merges two completed-commit lists, dedupes by requested time, and orders by completion time. + * 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 dedupedByRequestedTime = new LinkedHashMap<>(); + Map dedupedByRequestedTimeAndAction = new LinkedHashMap<>(); Stream.concat(list1.stream(), list2.stream()) .forEach( hoodieInstant -> - dedupedByRequestedTime.putIfAbsent(hoodieInstant.requestedTime(), hoodieInstant)); - return dedupedByRequestedTime.values().stream() + dedupedByRequestedTimeAndAction.putIfAbsent( + hoodieInstant.requestedTime() + "_" + hoodieInstant.getAction(), + hoodieInstant)); + return dedupedByRequestedTimeAndAction.values().stream() .sorted(Comparator.comparing(HoodieInstant::getCompletionTime)) .collect(Collectors.toList()); } From b87173796aeaa9d63604621c28934e2f7ef2e2b7 Mon Sep 17 00:00:00 2001 From: Vinish Reddy Date: Wed, 1 Jul 2026 16:28:53 -0700 Subject: [PATCH 6/8] Apply spotless formatting --- .../org/apache/xtable/hudi/BaseFileUpdatesExtractor.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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 06c999a2e..1ad8bb1c2 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 @@ -80,7 +80,8 @@ public class BaseFileUpdatesExtractor { // 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 static final Pattern EXTERNAL_FILE_GROUP_PREFIX_PATTERN = + Pattern.compile("bucket-[0-9]+"); private final HoodieEngineContext engineContext; private final Path tableBasePath; @@ -332,7 +333,8 @@ private WriteStatus toWriteStatus( fileName, commitTime, prefix)) .orElseGet( () -> - ExternalFilePathUtil.appendCommitTimeAndExternalFileMarker(filePath, commitTime)); + ExternalFilePathUtil.appendCommitTimeAndExternalFileMarker( + filePath, commitTime)); writeStatus.setFileId(fileId); writeStatus.setPartitionPath(partitionPath); HoodieDeltaWriteStat writeStat = new HoodieDeltaWriteStat(); From 994aa31ddf9085442e13e7dddd71fb83693c6a63 Mon Sep 17 00:00:00 2001 From: Vinish Reddy Date: Wed, 1 Jul 2026 16:28:53 -0700 Subject: [PATCH 7/8] Default the Hudi target to table version 6 and parameterize tests for versions 6 and 9 Flip HudiTargetConfig.DEFAULT_TABLE_VERSION from NINE to SIX so the default output stays readable by released Hudi readers; version 9 remains fully supported via xtable.hudi.target.table_version=9. Parameterize the Hudi test suites so every run exercises both table versions instead of only the default: - ITHudiConversionTarget: partitioned x {SIX, NINE} via a MethodSource cross-product; the target client sets the version through HudiTargetConfig.HUDI_TABLE_VERSION. - ITHudiConversionSource: source tables are created at {SIX, NINE} via the table-type/partition MethodSource cross-products, and the parameterized tests write through TestSparkHudiTable (Spark writer) instead of the Java client. - ITConversionController: combinations targeting HUDI are emitted once per version; getTableSyncConfig gained an overload that applies the version to the Hudi target properties. - TestHudiTargetConfig/TestHudiConversionTarget assert against DEFAULT_TABLE_VERSION instead of a hard-coded version. Version 9 source coverage in ITHudiConversionSource depends on two Hudi fixes validated against a locally patched 1.3.0-SNAPSHOT: apache/hudi#19126 (column stats on map/array-nested leaves during MOR log-append) and the savepoint backlog fix in the previous commit. --- .../apache/xtable/hudi/HudiTargetConfig.java | 2 +- .../apache/xtable/ITConversionController.java | 101 ++++++++++++++---- .../org/apache/xtable/TestSparkHudiTable.java | 25 +++++ .../xtable/hudi/ITHudiConversionSource.java | 87 +++++++++------ .../xtable/hudi/ITHudiConversionTarget.java | 57 ++++++---- .../xtable/hudi/TestHudiConversionTarget.java | 3 +- .../xtable/hudi/TestHudiTargetConfig.java | 6 +- 7 files changed, 207 insertions(+), 74 deletions(-) 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 index 3d3131ea2..3e2a4dc37 100644 --- a/xtable-core/src/main/java/org/apache/xtable/hudi/HudiTargetConfig.java +++ b/xtable-core/src/main/java/org/apache/xtable/hudi/HudiTargetConfig.java @@ -34,7 +34,7 @@ public class HudiTargetConfig { */ public static final String HUDI_TABLE_VERSION = "xtable.hudi.target.table_version"; - static final HoodieTableVersion DEFAULT_TABLE_VERSION = HoodieTableVersion.NINE; + static final HoodieTableVersion DEFAULT_TABLE_VERSION = HoodieTableVersion.SIX; HoodieTableVersion tableVersion; 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 6b0a1b210..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,7 +236,10 @@ 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); String partitionConfig = null; @@ -248,7 +262,8 @@ public void testVariousOperations( table, targetTableFormats, partitionConfig, - null); + null, + hudiTargetVersion); conversionController.sync(conversionConfig, conversionSourceProvider); checkDatasetEquivalence(sourceTableFormat, table, targetTableFormats, 100); @@ -280,7 +295,8 @@ public void testVariousOperations( tableWithUpdatedSchema, targetTableFormats, partitionConfig, - null); + null, + hudiTargetVersion); List insertsAfterSchemaUpdate = tableWithUpdatedSchema.insertRows(100); tableWithUpdatedSchema.reload(); conversionController.sync(conversionConfig, conversionSourceProvider); @@ -524,43 +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)), - Arguments.of( + DELTA, Arrays.asList(ICEBERG, HUDI), null, "level:VALUE", levelFilter), buildArgsForPartition( - ICEBERG, Arrays.asList(DELTA, HUDI), null, "level:VALUE", levelFilter)), - // Delta is excluded here since it does not support nested partition columns. - Arguments.of( + ICEBERG, Arrays.asList(DELTA, HUDI), null, "level:VALUE", levelFilter), + // Delta is excluded here since it does not support nested partition columns. buildArgsForPartition( HUDI, Arrays.asList(ICEBERG), "nested_record.level:SIMPLE", "nested_record.level:VALUE", - nestedLevelFilter)), - Arguments.of( + 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(); @@ -587,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 @@ -1184,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); @@ -1207,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()); @@ -1217,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/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/ITHudiConversionSource.java b/xtable-core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSource.java index 979a4a636..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 @@ -234,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<>(); @@ -381,9 +383,9 @@ 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<>(); @@ -429,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); @@ -465,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<>(); @@ -514,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<>(); @@ -562,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<>(); @@ -615,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<>(); @@ -682,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<>(); @@ -749,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); @@ -810,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() { @@ -822,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 f75d4dd79..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); @@ -240,8 +244,8 @@ void syncForExistingTable(boolean partitioned) { } @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); @@ -258,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 = @@ -285,8 +289,11 @@ void syncForNewTable(boolean partitioned) { } @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); @@ -309,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 = @@ -422,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"; @@ -446,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 = @@ -512,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"; @@ -534,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); @@ -775,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/TestHudiConversionTarget.java b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiConversionTarget.java index a2a24a812..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, HoodieTableVersion.NINE)) + 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/TestHudiTargetConfig.java b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiTargetConfig.java index 93a2f4b8b..6cd10150c 100644 --- a/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiTargetConfig.java +++ b/xtable-core/src/test/java/org/apache/xtable/hudi/TestHudiTargetConfig.java @@ -30,11 +30,11 @@ public class TestHudiTargetConfig { @Test - void defaultsToTableVersionNine() { + void defaultsToTableVersionSix() { assertEquals( - HoodieTableVersion.NINE, + HoodieTableVersion.SIX, HudiTargetConfig.fromProperties(new Properties()).getTableVersion()); - assertEquals(HoodieTableVersion.NINE, HudiTargetConfig.fromProperties(null).getTableVersion()); + assertEquals(HoodieTableVersion.SIX, HudiTargetConfig.fromProperties(null).getTableVersion()); } @Test From 5aca977ab4e5971501d920e449fa2b57d7f0239e Mon Sep 17 00:00:00 2001 From: Vinish Reddy Date: Mon, 17 Aug 2026 14:13:34 -0700 Subject: [PATCH 8/8] [834] Follow Hudi's hudi-common package reorganization Hudi master moved two packages that XTable depends on, in the refactor(common) commits apache/hudi#19193 and apache/hudi#19195: - org.apache.hudi.stats -> org.apache.hudi.metadata.stats, covering ValueMetadata, ValueType and HoodieColumnRangeMetadata. XTable's XTableValueMetadata and its test live inside Hudi's package to reach package-private members, so the files move directories as well. - org.apache.hudi.avro.HoodieAvroUtils -> org.apache.hudi.common.avro. HoodieAvroUtils. The generated org.apache.hudi.avro.model classes did not move. This is required to build against a Hudi release that carries the reorganization, alongside the hudi.version bump this branch already makes. --- .../hudi/{ => metadata}/stats/XTableValueMetadata.java | 2 +- .../org/apache/xtable/hudi/BaseFileUpdatesExtractor.java | 6 +++--- .../org/apache/xtable/hudi/HudiFileStatsExtractor.java | 6 +++--- .../{ => metadata}/stats/TestXTableValueMetadata.java | 2 +- .../test/java/org/apache/xtable/TestJavaHudiTable.java | 2 +- .../test/java/org/apache/xtable/hudi/HudiTestUtil.java | 2 +- .../apache/xtable/hudi/TestBaseFileUpdatesExtractor.java | 8 ++++---- 7 files changed, 14 insertions(+), 14 deletions(-) rename xtable-core/src/main/java/org/apache/hudi/{ => metadata}/stats/XTableValueMetadata.java (99%) rename xtable-core/src/test/java/org/apache/hudi/{ => metadata}/stats/TestXTableValueMetadata.java (99%) 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 1ad8bb1c2..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; 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/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/TestJavaHudiTable.java b/xtable-core/src/test/java/org/apache/xtable/TestJavaHudiTable.java index 8e8d05c0c..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; 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/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;