This is an automated email from the ASF dual-hosted git repository.
vinishjail97 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
The following commit(s) were added to refs/heads/main by this push:
new 788b9ae7 [834] Build the column-stats index for existing Hudi targets
(#961)
788b9ae7 is described below
commit 788b9ae7b5e921f2722c6a37845e40ce6c0f06d4
Author: Vinish Reddy <[email protected]>
AuthorDate: Wed Oct 7 06:53:51 2026 +0530
[834] Build the column-stats index for existing Hudi targets (#961)
- Require a snapshot sync when a Hudi target has commits but no
column-stats index
- Register every snapshot file again in that sync, so Hudi indexes all
files from the
commit's write stats; Hudi's own bootstrap skips externally created files
- Never replace a fileId that the same commit writes
- Find older file slices of live file groups when marking files as cleaned
---
.../apache/xtable/spi/sync/ConversionTarget.java | 12 ++
.../xtable/conversion/ConversionController.java | 6 +
.../xtable/hudi/BaseFileUpdatesExtractor.java | 37 +++-
.../apache/xtable/hudi/HudiConversionTarget.java | 68 ++++--
.../conversion/TestConversionController.java | 56 +++++
.../apache/xtable/hudi/ITHudiConversionTarget.java | 227 +++++++++++++++++++++
6 files changed, 380 insertions(+), 26 deletions(-)
diff --git
a/xtable-api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java
b/xtable-api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java
index 6f8f9395..fda3e7b5 100644
--- a/xtable-api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java
+++ b/xtable-api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java
@@ -87,6 +87,18 @@ public interface ConversionTarget {
/** Returns the xtable metadata persisted in the target */
Optional<TableSyncMetadata> getTableMetadata();
+ /**
+ * Checks if an incremental sync can bring the target table up to date. The
source decides this
+ * with {@link
org.apache.xtable.spi.extractor.ConversionSource#isIncrementalSyncSafeFrom};
this
+ * method lets the target also require a snapshot sync, for example when the
target is missing
+ * metadata that only the full set of source files can build.
+ *
+ * @return true if incremental sync is safe for the target table, false
otherwise
+ */
+ default boolean isIncrementalSyncSafe() {
+ return true;
+ }
+
/** Returns the TableFormat name the client syncs to */
String getTableFormat();
diff --git
a/xtable-core/src/main/java/org/apache/xtable/conversion/ConversionController.java
b/xtable-core/src/main/java/org/apache/xtable/conversion/ConversionController.java
index 73bc531a..8bbc147d 100644
---
a/xtable-core/src/main/java/org/apache/xtable/conversion/ConversionController.java
+++
b/xtable-core/src/main/java/org/apache/xtable/conversion/ConversionController.java
@@ -263,6 +263,12 @@ public class ConversionController {
return conversionTargetByFormat.entrySet().stream()
.filter(
entry -> {
+ if (!entry.getValue().isIncrementalSyncSafe()) {
+ log.info(
+ "Incremental sync is not safe for target {}. Falling back
to snapshot sync.",
+ entry.getKey());
+ return false;
+ }
Optional<Instant> lastSyncInstant =
lastSyncMetadataByFormat
.get(entry.getKey())
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 349b8daa..2bd1134b 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,6 +57,7 @@ import org.apache.hudi.common.util.ExternalFilePathUtil;
import org.apache.hudi.hadoop.fs.CachingPath;
import org.apache.hudi.metadata.HoodieIndexVersion;
import org.apache.hudi.metadata.HoodieTableMetadata;
+import org.apache.hudi.metadata.MetadataPartitionType;
import org.apache.hudi.stats.HoodieColumnRangeMetadata;
import org.apache.hudi.stats.ValueMetadata;
import org.apache.hudi.stats.XTableValueMetadata;
@@ -141,6 +142,14 @@ public class BaseFileUpdatesExtractor {
HoodieTableMetaClient metaClient,
HoodieMetadataConfig metadataConfig) {
boolean isTableInitialized = metaClient.isTimelineNonEmpty();
+ // Hudi builds the column-stats index for existing files from the metadata
table's file listing,
+ // which skips externally created files. When the index is missing,
register every snapshot file
+ // again, so the index gets the stats of all files from this commit's
write stats.
+ boolean reregisterExistingFiles =
+ isTableInitialized
+ && !metaClient
+ .getTableConfig()
+
.isMetadataPartitionAvailable(MetadataPartitionType.COLUMN_STATS);
// Track the partitions that are not present in the snapshot, so the files
for those partitions
// can be dropped
HoodieIndexVersion indexVersion =
@@ -169,18 +178,15 @@ public class BaseFileUpdatesExtractor {
: Collections.emptyList();
Set<String> existingPaths =
baseFiles.stream().map(HoodieBaseFile::getPath).collect(Collectors.toSet());
- // Mark fileIds for removal if the file paths are no longer
present in the
- // snapshot
- List<String> fileIdsToRemove =
- baseFiles.stream()
- .filter(baseFile ->
!physicalPathToFile.containsKey(baseFile.getPath()))
- .map(HoodieBaseFile::getFileId)
- .collect(Collectors.toList());
// for any entries in the map that are not in the set of
existing paths, create a
- // write status to add them to the Hudi table
+ // write status to add them to the Hudi table; an existing
file gets a new file
+ // slice in its file group when it is registered again
List<WriteStatus> writeStatuses =
physicalPathToFile.entrySet().stream()
- .filter(entry ->
!existingPaths.contains(entry.getKey()))
+ .filter(
+ entry ->
+ reregisterExistingFiles
+ ||
!existingPaths.contains(entry.getKey()))
.map(Map.Entry::getValue)
.map(
snapshotFile ->
@@ -191,6 +197,19 @@ public class BaseFileUpdatesExtractor {
Optional.of(partitionPath),
indexVersion))
.collect(Collectors.toList());
+ // Mark fileIds for removal if the file paths are no longer
present in the
+ // snapshot. A fileId that this commit writes is never
replaced, because Hudi
+ // hides every file slice of a replaced file group.
+ Set<String> writtenFileIds =
+ writeStatuses.stream()
+ .map(WriteStatus::getFileId)
+ .collect(Collectors.toSet());
+ List<String> fileIdsToRemove =
+ baseFiles.stream()
+ .filter(baseFile ->
!physicalPathToFile.containsKey(baseFile.getPath()))
+ .map(HoodieBaseFile::getFileId)
+ .filter(fileId -> !writtenFileIds.contains(fileId))
+ .collect(Collectors.toList());
return ReplaceMetadata.of(
fileIdsToRemove.isEmpty()
? Collections.emptyMap()
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 6523b519..a463062c 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
@@ -33,7 +33,6 @@ import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
-import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -81,6 +80,7 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.hadoop.fs.CachingPath;
import org.apache.hudi.metadata.HoodieIndexVersion;
import org.apache.hudi.metadata.HoodieTableMetadataWriter;
+import org.apache.hudi.metadata.MetadataPartitionType;
import org.apache.hudi.table.HoodieJavaTable;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.clean.CleanPlanner;
@@ -403,6 +403,30 @@ public class HudiConversionTarget implements
ConversionTarget {
.flatMap(instant -> getMetadata(instant, client)));
}
+ /**
+ * XTable 0.4.0 did not build the column-stats index for partitioned tables.
Hudi cannot build it
+ * later for the externally created files, so a snapshot sync registers
every file again with its
+ * stats.
+ */
+ @Override
+ public boolean isIncrementalSyncSafe() {
+ return metaClient
+ .map(
+ client -> {
+ if (!client.isTimelineNonEmpty()
+ || client
+ .getTableConfig()
+
.isMetadataPartitionAvailable(MetadataPartitionType.COLUMN_STATS)) {
+ return true;
+ }
+ log.info(
+ "Hudi target table at {} has no column-stats index, so a
snapshot sync builds it.",
+ tableDataPath);
+ return false;
+ })
+ .orElse(true);
+ }
+
@Override
public String getTableFormat() {
return TableFormat.HUDI;
@@ -590,23 +614,33 @@ public class HudiConversionTarget implements
ConversionTarget {
deletePathsForPartition -> {
String partition = deletePathsForPartition.getKey();
// we need to manipulate the path to properly clean
from the metadata table,
- // so we map the file path to the base file
- Map<String, HoodieBaseFile> baseFilesByPath =
- fsView
- .getAllReplacedFileGroups(partition)
- .flatMap(HoodieFileGroup::getAllBaseFiles)
- .collect(
- Collectors.toMap(HoodieBaseFile::getPath,
Function.identity()));
+ // so we map the file path to the base files that can
be cleaned. A file
+ // registered again by a snapshot sync has an older
file slice in a live
+ // file group with the same path, so the latest slice
of a live file group
+ // is never a candidate.
+ Map<String, List<HoodieBaseFile>>
cleanableBaseFilesByPath =
+ Stream.concat(
+ fsView
+ .getAllFileGroups(partition)
+ .flatMap(fileGroup ->
fileGroup.getAllBaseFiles().skip(1)),
+ fsView
+ .getAllReplacedFileGroups(partition)
+
.flatMap(HoodieFileGroup::getAllBaseFiles))
+
.collect(Collectors.groupingBy(HoodieBaseFile::getPath));
return
deletePathsForPartition.getValue().getValue().stream()
+ .map(cleanFileInfo -> cleanFileInfo.getFilePath())
+ .distinct()
+ .flatMap(
+ path ->
+ cleanableBaseFilesByPath
+ .getOrDefault(path,
Collections.emptyList())
+ .stream())
.map(
- cleanFileInfo -> {
- HoodieBaseFile baseFile =
-
baseFilesByPath.get(cleanFileInfo.getFilePath());
- return new HoodieCleanFileInfo(
-
ExternalFilePathUtil.appendCommitTimeAndExternalFileMarker(
- baseFile.getFileName(),
baseFile.getCommitTime()),
- false);
- })
+ baseFile ->
+ new HoodieCleanFileInfo(
+
ExternalFilePathUtil.appendCommitTimeAndExternalFileMarker(
+ baseFile.getFileName(),
baseFile.getCommitTime()),
+ false))
.collect(Collectors.toList());
}));
// there is nothing to clean, so exit early
@@ -656,7 +690,7 @@ public class HudiConversionTarget implements
ConversionTarget {
String partitionPath = entry.getKey();
List<String> deletePaths =
entry.getValue().stream()
- .map(HoodieCleanFileInfo::getFilePath)
+ .map(cleanFileInfo ->
cleanFileInfo.getFilePath())
.collect(Collectors.toList());
return HoodieCleanStat.builder()
.withPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS)
diff --git
a/xtable-core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
b/xtable-core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
index 6569c4e8..3d7c8808 100644
---
a/xtable-core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
+++
b/xtable-core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
@@ -96,6 +96,8 @@ public class TestConversionController {
void setupTargetFormats() {
when(mockConversionTarget1.getTableFormat()).thenReturn(ICEBERG);
when(mockConversionTarget2.getTableFormat()).thenReturn(DELTA);
+ when(mockConversionTarget1.isIncrementalSyncSafe()).thenReturn(true);
+ when(mockConversionTarget2.isIncrementalSyncSafe()).thenReturn(true);
}
@Test
@@ -285,6 +287,60 @@ public class TestConversionController {
assertEquals(syncResults, result);
}
+ @Test
+ void testIncrementalSyncFallBackToSnapshotWhenTargetIsNotSafe() {
+ SyncMode syncMode = SyncMode.INCREMENTAL;
+ InternalTable internalTable = getInternalTable();
+ Instant instantBeforeHour = Instant.now().minus(Duration.ofHours(1));
+ InternalSnapshot internalSnapshot = buildSnapshot(internalTable, "v1",
"0");
+ SyncResult syncResult = buildSyncResult(syncMode, instantBeforeHour);
+ Map<String, SyncResult> syncResults = new HashMap<>();
+ syncResults.put(TableFormat.ICEBERG, syncResult);
+ syncResults.put(TableFormat.DELTA, syncResult);
+ ConversionConfig conversionConfig =
+ getTableSyncConfig(Arrays.asList(TableFormat.ICEBERG,
TableFormat.DELTA), syncMode);
+ when(mockConversionSourceProvider.getConversionSourceInstance(
+ conversionConfig.getSourceTable()))
+ .thenReturn(mockConversionSource);
+ when(mockConversionTargetFactory.createForFormat(
+ conversionConfig.getTargetTables().get(0), mockConf))
+ .thenReturn(mockConversionTarget1);
+ when(mockConversionTargetFactory.createForFormat(
+ conversionConfig.getTargetTables().get(1), mockConf))
+ .thenReturn(mockConversionTarget2);
+
+ Instant instantAsOfNow = Instant.now();
+ Instant instantAt5 = getInstantAtLastNMinutes(instantAsOfNow, 5);
+
when(mockConversionSource.isIncrementalSyncSafeFrom(eq(instantAt5))).thenReturn(true);
+ // The source allows an incremental sync, but both targets require a
snapshot sync.
+ when(mockConversionTarget1.isIncrementalSyncSafe()).thenReturn(false);
+ when(mockConversionTarget2.isIncrementalSyncSafe()).thenReturn(false);
+
+ // Both Iceberg and Delta last synced at instantAt5 and have no pending
instants.
+ when(mockConversionTarget1.getTableMetadata())
+ .thenReturn(
+ Optional.of(TableSyncMetadata.of(instantAt5,
Collections.emptyList(), "TEST", "0")));
+ when(mockConversionTarget2.getTableMetadata())
+ .thenReturn(
+ Optional.of(TableSyncMetadata.of(instantAt5,
Collections.emptyList(), "TEST", "0")));
+
+
when(mockConversionSource.getCurrentSnapshot()).thenReturn(internalSnapshot);
+ when(tableFormatSync.syncSnapshot(
+ argThat(containsAll(Arrays.asList(mockConversionTarget1,
mockConversionTarget2))),
+ eq(internalSnapshot)))
+ .thenReturn(syncResults);
+ ConversionController conversionController =
+ new ConversionController(
+ mockConf,
+ mockConversionTargetFactory,
+ mockCatalogConversionFactory,
+ tableFormatSync,
+ catalogSync);
+ Map<String, SyncResult> result =
+ conversionController.sync(conversionConfig,
mockConversionSourceProvider);
+ assertEquals(syncResults, result);
+ }
+
@Test
void testIncrementalSyncFallbackToSnapshotForOnlySingleFormat() {
SyncMode syncMode = SyncMode.INCREMENTAL;
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 64d5e9ca..0707f435 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
@@ -27,6 +27,8 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.parallel.ExecutionMode.SAME_THREAD;
+import java.io.IOException;
+import java.io.UncheckedIOException;
import java.nio.file.Path;
import java.time.Duration;
import java.time.Instant;
@@ -46,7 +48,12 @@ import lombok.SneakyThrows;
import org.apache.avro.Schema;
import org.apache.avro.SchemaBuilder;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.avro.AvroParquetWriter;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.util.HadoopOutputFile;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.io.TempDir;
@@ -61,6 +68,7 @@ import org.apache.hudi.avro.model.StringWrapper;
import org.apache.hudi.client.HoodieJavaWriteClient;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.common.HoodieJavaEngineContext;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieBaseFile;
@@ -80,6 +88,7 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.metadata.HoodieBackedTableMetadata;
+import org.apache.hudi.metadata.MetadataPartitionType;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.xtable.conversion.TargetTable;
@@ -246,6 +255,224 @@ public class ITHudiConversionTarget {
assertEquals(tableVersion, metaClient.getTableConfig().getTableVersion());
}
+ /**
+ * XTable 0.4.0 did not build the column-stats index for partitioned
targets, and Hudi cannot
+ * build it later for externally created files. The target must require one
snapshot sync, which
+ * registers every file again with its stats, and a later clean must keep
those stats.
+ */
+ @ParameterizedTest
+ @EnumSource(
+ value = HoodieTableVersion.class,
+ names = {"SIX", "NINE"})
+ void
snapshotSyncBuildsColStatsForExistingTableWithoutColStats(HoodieTableVersion
tableVersion) {
+ String partitionPath = "partition_path";
+ String existingFileName = "existing_file_1.parquet";
+ String existingFilePath = getFilePath(partitionPath, existingFileName);
+ HoodieTableMetaClient setupMetaClient =
+ initTableAndGetMetaClient(tableBasePath, PARTITION_FIELD_NAME,
tableVersion);
+ Schema avroSchema =
+ SchemaBuilder.record(TEST_SCHEMA_NAME)
+ .fields()
+ .requiredString(KEY_FIELD_NAME)
+ .requiredString(PARTITION_FIELD_NAME)
+ .requiredString(OTHER_FIELD_NAME)
+ .endRecord();
+ writeParquetFile(tableBasePath + "/" + existingFilePath, avroSchema);
+ // register the file the way XTable 0.4.0 did, without the column-stats
index
+ HoodieWriteConfig baseWriteConfig = getHoodieWriteConfig(setupMetaClient,
avroSchema);
+ HoodieWriteConfig writeConfigWithoutColStats =
+ HoodieWriteConfig.newBuilder()
+ .withProperties(baseWriteConfig.getProps())
+ .withMetadataConfig(
+ HoodieMetadataConfig.newBuilder()
+
.fromProperties(baseWriteConfig.getMetadataConfig().getProps())
+ .withMetadataIndexColumnStats(false)
+ .build())
+ .build();
+ String initialInstant =
+ HudiInstantUtils.convertInstantToCommit(Instant.now().minus(30,
ChronoUnit.HOURS));
+ try (HoodieJavaWriteClient<?> writeClient =
+ new HoodieJavaWriteClient<>(CONTEXT, writeConfigWithoutColStats)) {
+ setupMetaClient
+ .getActiveTimeline()
+ .createRequestedCommitWithReplaceMetadata(
+ initialInstant, HoodieTimeline.REPLACE_COMMIT_ACTION);
+ setupMetaClient
+ .getActiveTimeline()
+ .transitionReplaceRequestedToInflight(
+ new HoodieInstant(
+ HoodieInstant.State.REQUESTED,
+ HoodieTimeline.REPLACE_COMMIT_ACTION,
+ initialInstant,
+ InstantComparatorV2.REQUESTED_TIME_BASED_COMPARATOR),
+ Option.empty());
+ writeClient.setOperationType(WriteOperationType.INSERT);
+ writeClient.commit(
+ initialInstant,
+ Collections.singletonList(
+ createWriteStatus(
+ existingFileName, partitionPath, initialInstant, 2, 100,
Collections.emptyMap())),
+ Option.empty(),
+ HoodieTimeline.REPLACE_COMMIT_ACTION,
+ Collections.emptyMap());
+ }
+ assertFalse(
+ HoodieTableMetaClient.reload(setupMetaClient)
+ .getTableConfig()
+ .isMetadataPartitionAvailable(MetadataPartitionType.COLUMN_STATS));
+ assertFalse(getTargetClient(tableVersion).isIncrementalSyncSafe());
+
+ // the snapshot sync registers the existing file again and adds a new file
+ String fileName = "file_1.parquet";
+ String filePath = getFilePath(partitionPath, fileName);
+ InternalTable state = getState(Instant.now().minus(20, ChronoUnit.HOURS),
true);
+ HudiConversionTarget targetClient = getTargetClient(tableVersion);
+ targetClient.beginSync(state);
+ targetClient.syncFilesForSnapshot(
+ Collections.singletonList(
+ PartitionFileGroup.builder()
+ .files(
+ Arrays.asList(
+ getExistingTestFile(partitionPath, existingFileName),
+ getTestFile(partitionPath, fileName)))
+ .partitionValues(
+ Collections.singletonList(
+ PartitionValue.builder()
+ .partitionField(PARTITION_FIELD)
+ .range(Range.scalar("partitionPath"))
+ .build()))
+ .build()));
+ targetClient.syncSchema(SCHEMA);
+ targetClient.syncMetadata(
+ TableSyncMetadata.of(state.getLatestCommitTime(),
Collections.emptyList(), "TEST", "0"));
+ targetClient.completeSync();
+
+ HoodieTableMetaClient metaClient =
+
HoodieTableMetaClient.builder().setConf(CONFIGURATION).setBasePath(tableBasePath).build();
+ assertTrue(
+ metaClient
+ .getTableConfig()
+ .isMetadataPartitionAvailable(MetadataPartitionType.COLUMN_STATS));
+ assertTrue(getTargetClient(tableVersion).isIncrementalSyncSafe());
+ // the existing file keeps its physical path
+ List<Pair<String, String>> expectedFiles =
+ Arrays.asList(Pair.of(existingFileName, existingFilePath),
Pair.of(fileName, filePath));
+ assertFileGroupCorrectness(metaClient, partitionPath, expectedFiles);
+ assertColStatsForExistingAndNewFile(
+ metaClient, baseWriteConfig, partitionPath, existingFileName,
fileName);
+
+ // the cleaner keeps the commits of the last 4 hours, so a commit 1 hour
ago makes the next
+ // commit clean the older file slice of the existing file
+ incrementalSync(
+ targetClient,
+ Collections.singletonList(getTestFile(partitionPath,
"file_2.parquet")),
+ Collections.emptyList(),
+ Instant.now().minus(1, ChronoUnit.HOURS),
+ "1",
+ true);
+ incrementalSync(
+ targetClient,
+ Collections.singletonList(getTestFile(partitionPath,
"file_3.parquet")),
+ Collections.emptyList(),
+ Instant.now(),
+ "2",
+ true);
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ assertTrue(
+ metaClient
+ .getActiveTimeline()
+ .getCleanerTimeline()
+ .filterCompletedInstants()
+ .countInstants()
+ > 0);
+ assertColStatsForExistingAndNewFile(
+ metaClient, baseWriteConfig, partitionPath, existingFileName,
fileName);
+ try (HoodieTableFileSystemView fsView =
+ new HoodieTableFileSystemView(
+ new HoodieBackedTableMetadata(
+ CONTEXT,
+ metaClient.getStorage(),
+ baseWriteConfig.getMetadataConfig(),
+ tableBasePath,
+ true),
+ metaClient,
+ metaClient.getActiveTimeline())) {
+ // the clean removed the older file slice and kept the one from the
snapshot sync
+ HoodieFileGroup existingFileGroup =
+ fsView
+ .getAllFileGroups(partitionPath)
+ .filter(fileGroup ->
fileGroup.getFileGroupId().getFileId().equals(existingFileName))
+ .findFirst()
+ .get();
+ List<HoodieBaseFile> existingBaseFiles =
+ existingFileGroup.getAllBaseFiles().collect(Collectors.toList());
+ assertEquals(1, existingBaseFiles.size());
+ assertEquals(
+ HudiInstantUtils.convertInstantToCommit(state.getLatestCommitTime()),
+ existingBaseFiles.get(0).getCommitTime());
+ assertEquals(tableBasePath + "/" + existingFilePath,
existingBaseFiles.get(0).getPath());
+ }
+ }
+
+ private void assertColStatsForExistingAndNewFile(
+ HoodieTableMetaClient metaClient,
+ HoodieWriteConfig writeConfig,
+ String partitionPath,
+ String existingFileName,
+ String fileName) {
+ try (HoodieBackedTableMetadata hoodieBackedTableMetadata =
+ new HoodieBackedTableMetadata(
+ CONTEXT,
+ metaClient.getStorage(),
+ writeConfig.getMetadataConfig(),
+ tableBasePath,
+ true)) {
+ assertColStats(hoodieBackedTableMetadata, partitionPath, fileName);
+ assertColStatsForField(
+ hoodieBackedTableMetadata,
+ partitionPath,
+ existingFileName,
+ KEY_FIELD_NAME,
+ "id3",
+ "id4",
+ 2,
+ 0,
+ 5);
+ }
+ }
+
+ private InternalDataFile getExistingTestFile(String partitionPath, String
fileName) {
+ return getTestFile(partitionPath, fileName).toBuilder()
+ .columnStats(
+ Collections.singletonList(
+ ColumnStat.builder()
+ .field(SCHEMA.getFields().get(0))
+ .range(Range.vector("id3", "id4"))
+ .numNulls(0)
+ .numValues(2)
+ .totalSize(5)
+ .build()))
+ .build();
+ }
+
+ private static void writeParquetFile(String path, Schema avroSchema) {
+ try (ParquetWriter<GenericRecord> writer =
+ AvroParquetWriter.<GenericRecord>builder(
+ HadoopOutputFile.fromPath(new org.apache.hadoop.fs.Path(path),
new Configuration()))
+ .withSchema(avroSchema)
+ .build()) {
+ for (String id : Arrays.asList("id3", "id4")) {
+ GenericRecord record = new GenericData.Record(avroSchema);
+ record.put(KEY_FIELD_NAME, id);
+ record.put(PARTITION_FIELD_NAME, "c");
+ record.put(OTHER_FIELD_NAME, "content");
+ writer.write(record);
+ }
+ } catch (IOException ex) {
+ throw new UncheckedIOException(ex);
+ }
+ }
+
@ParameterizedTest
@MethodSource("partitionedAndVersion")
void syncForNewTable(boolean partitioned, HoodieTableVersion tableVersion) {
