Skip to content

[Hudi-3376] Add an option to skip under deletion files for HoodieMetadataTableValidator #4994

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yihua merged 5 commits into from
Mar 17, 2022
Merged

[Hudi-3376] Add an option to skip under deletion files for HoodieMetadataTableValidator #4994

Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
eb7d833
ready to review
Mar 8, 2022
9225487
tested and ready for review
Mar 9, 2022
95d3fa0
fix code style
Mar 9, 2022
9c6b594
merge from master
Mar 10, 2022
56940f0
code review
Mar 14, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
192 changes: 162 additions & 30 deletions hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.apache.hudi.utilities;

import org.apache.hudi.async.HoodieAsyncService;
import org.apache.hudi.avro.model.HoodieCleanerPlan;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.common.bloom.BloomFilter;
import org.apache.hudi.common.config.HoodieMetadataConfig;
Expand All @@ -29,16 +30,21 @@
import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieColumnRangeMetadata;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieFileGroup;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.TableSchemaResolver;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.view.FileSystemViewManager;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
import org.apache.hudi.common.util.CleanerUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ParquetUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieValidationException;
import org.apache.hudi.io.storage.HoodieFileReader;
import org.apache.hudi.io.storage.HoodieFileReaderFactory;
Expand All @@ -55,6 +61,7 @@
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
Expand All @@ -67,6 +74,7 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
* A validator with spark-submit to compare information, such as partitions, file listing, index, etc.,
Expand Down Expand Up @@ -173,6 +181,9 @@ public static class Config implements Serializable {
+ "Can use --min-validate-interval-seconds to control validation frequency", required = false)
public boolean continuous = false;

@Parameter(names = {"--skip-under-deletion-data-files"}, description = "Skip to compare the data files which are under deletion by cleaner", required = false)
Comment thread
yihua marked this conversation as resolved.
Outdated
public boolean skipUnderDeletionDataFiles = false;

@Parameter(names = {"--validate-latest-file-slices"}, description = "Validate latest file slices for all partitions.", required = false)
public boolean validateLatestFileSlices = false;

Expand Down Expand Up @@ -230,6 +241,7 @@ public String toString() {
+ " --validate-all-column-stats " + validateAllColumnStats + ", \n"
+ " --validate-bloom-filters " + validateBloomFilters + ", \n"
+ " --continuous " + continuous + ", \n"
+ " --skip-under-deletion-data-files " + skipUnderDeletionDataFiles + ", \n"
+ " --ignore-failed " + ignoreFailed + ", \n"
+ " --min-validate-interval-seconds " + minValidateIntervalSeconds + ", \n"
+ " --parallelism " + parallelism + ", \n"
Expand All @@ -252,6 +264,7 @@ public boolean equals(Object o) {
Config config = (Config) o;
return basePath.equals(config.basePath)
&& Objects.equals(continuous, config.continuous)
&& Objects.equals(skipUnderDeletionDataFiles, config.skipUnderDeletionDataFiles)
&& Objects.equals(validateLatestFileSlices, config.validateLatestFileSlices)
&& Objects.equals(validateLatestBaseFiles, config.validateLatestBaseFiles)
&& Objects.equals(validateAllFileGroups, config.validateAllFileGroups)
Expand All @@ -269,7 +282,7 @@ public boolean equals(Object o) {

@Override
public int hashCode() {
return Objects.hash(basePath, continuous, validateLatestFileSlices, validateLatestBaseFiles,
return Objects.hash(basePath, continuous, skipUnderDeletionDataFiles, validateLatestFileSlices, validateLatestBaseFiles,
validateAllFileGroups, validateAllColumnStats, validateBloomFilters, minValidateIntervalSeconds,
parallelism, ignoreFailed, sparkMaster, sparkMemory, assumeDatePartitioning, propsFilePath, configs, help);
}
Expand Down Expand Up @@ -345,16 +358,59 @@ public void doMetadataTableValidation() {
boolean finalResult = true;
metaClient.reloadActiveTimeline();
String basePath = metaClient.getBasePath();
List<String> baseFilesUnderDeletion = Collections.emptyList();
Comment thread
yihua marked this conversation as resolved.
Outdated

if (cfg.skipUnderDeletionDataFiles) {
HoodieTimeline pendingCleaningTimeline = metaClient.getActiveTimeline()
.getCleanerTimeline()
.filter(instant -> instant.getState() != HoodieInstant.State.COMPLETED);
Comment thread
yihua marked this conversation as resolved.
Outdated

baseFilesUnderDeletion = pendingCleaningTimeline.getInstants().flatMap(instant -> {
try {
if (instant.isInflight()) {
// convert inflight instant to requested and get clean plan
instant = new HoodieInstant(HoodieInstant.State.REQUESTED, instant.getAction(), instant.getTimestamp());
}
HoodieCleanerPlan cleanerPlan = CleanerUtils.getCleanerPlan(metaClient, instant);

return cleanerPlan.getFilePathsToBeDeletedPerPartition().values().stream().flatMap(cleanerFIleInfoList -> {
Comment thread
yihua marked this conversation as resolved.
Outdated
return cleanerFIleInfoList.stream().map(fileInfo -> {
return new Path(fileInfo.getFilePath()).getName();
});
});

} catch (HoodieIOException ex) {

if (ex.getIOException() instanceof FileNotFoundException) {
// cleaner instant could be deleted by archive and FileNotFoundException could be threw during getInstantDetails function

@yihua yihua Mar 13, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Inflight cleaning instants shouldn't be archived. Is this considering the inflight cleaning instant deleted due to failed cleaning?

@zhangyue19921010 zhangyue19921010 Mar 14, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually no need to check FileNotFoundException here. removed.

// So that we need to catch the FileNotFoundException here and continue
LOG.warn(ex.getMessage());
return Stream.empty();
} else {
throw ex;
}

} catch (IOException e) {
throw new HoodieIOException("Error reading cleaner metadata for " + instant);
}
// only take care of base files here.
}).filter(path -> {
String fileExtension = FSUtils.getFileExtension(path);
return HoodieFileFormat.BASE_FILE_EXTENSIONS.contains(fileExtension);
}).collect(Collectors.toList());
}

HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc);
List<String> allPartitions = validatePartitions(engineContext, basePath);
HoodieMetadataValidationContext metadataTableBasedContext =
new HoodieMetadataValidationContext(engineContext, cfg, metaClient, true);
HoodieMetadataValidationContext fsBasedContext =
new HoodieMetadataValidationContext(engineContext, cfg, metaClient, false);

List<String> finalBaseFilesUnderDeletion = baseFilesUnderDeletion;
List<Boolean> result = engineContext.parallelize(allPartitions, allPartitions.size()).map(partitionPath -> {
try {
validateFilesInPartition(metadataTableBasedContext, fsBasedContext, partitionPath);
validateFilesInPartition(metadataTableBasedContext, fsBasedContext, partitionPath, finalBaseFilesUnderDeletion);
LOG.info("Metadata table validation succeeded for " + partitionPath);
return true;
} catch (HoodieValidationException e) {
Expand Down Expand Up @@ -410,42 +466,64 @@ private List<String> validatePartitions(HoodieSparkEngineContext engineContext,
* @param metadataTableBasedContext Validation context containing information based on metadata table
* @param fsBasedContext Validation context containing information based on the file system
* @param partitionPath Partition path String
* @param baseFilesUnderDeletion Base files under un-complete cleaner action
*/
private void validateFilesInPartition(
HoodieMetadataValidationContext metadataTableBasedContext,
HoodieMetadataValidationContext fsBasedContext, String partitionPath) {
HoodieMetadataValidationContext fsBasedContext, String partitionPath,
List<String> baseFilesUnderDeletion) {
if (cfg.validateLatestFileSlices) {
validateLatestFileSlices(metadataTableBasedContext, fsBasedContext, partitionPath);
validateLatestFileSlices(metadataTableBasedContext, fsBasedContext, partitionPath, baseFilesUnderDeletion);
}

if (cfg.validateLatestBaseFiles) {
validateLatestBaseFiles(metadataTableBasedContext, fsBasedContext, partitionPath);
validateLatestBaseFiles(metadataTableBasedContext, fsBasedContext, partitionPath, baseFilesUnderDeletion);
}

if (cfg.validateAllFileGroups) {
validateAllFileGroups(metadataTableBasedContext, fsBasedContext, partitionPath);
validateAllFileGroups(metadataTableBasedContext, fsBasedContext, partitionPath, baseFilesUnderDeletion);
}

if (cfg.validateAllColumnStats) {
validateAllColumnStats(metadataTableBasedContext, fsBasedContext, partitionPath);
validateAllColumnStats(metadataTableBasedContext, fsBasedContext, partitionPath, baseFilesUnderDeletion);
}

if (cfg.validateBloomFilters) {
validateBloomFilters(metadataTableBasedContext, fsBasedContext, partitionPath);
validateBloomFilters(metadataTableBasedContext, fsBasedContext, partitionPath, baseFilesUnderDeletion);
}
}

private void validateAllFileGroups(
HoodieMetadataValidationContext metadataTableBasedContext,
HoodieMetadataValidationContext fsBasedContext, String partitionPath) {
List<FileSlice> allFileSlicesFromMeta = metadataTableBasedContext
.getSortedAllFileGroupList(partitionPath).stream()
.flatMap(HoodieFileGroup::getAllFileSlices).sorted(new FileSliceComparator())
.collect(Collectors.toList());
List<FileSlice> allFileSlicesFromFS = fsBasedContext
.getSortedAllFileGroupList(partitionPath).stream()
.flatMap(HoodieFileGroup::getAllFileSlices).sorted(new FileSliceComparator())
.collect(Collectors.toList());
HoodieMetadataValidationContext fsBasedContext,
String partitionPath,
List<String> baseFilesUnderDeletion) {

@yihua yihua Mar 13, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

similar here for variable naming.

@zhangyue19921010 zhangyue19921010 Mar 14, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

chanegd.


List<FileSlice> allFileSlicesFromMeta;
List<FileSlice> allFileSlicesFromFS;

if (!baseFilesUnderDeletion.isEmpty()) {
List<FileSlice> fileSlicesFromMeta = metadataTableBasedContext
.getSortedAllFileGroupList(partitionPath).stream()
.flatMap(HoodieFileGroup::getAllFileSlices).sorted(new FileSliceComparator())
.collect(Collectors.toList());
List<FileSlice> fileSlicesFromFS = fsBasedContext
.getSortedAllFileGroupList(partitionPath).stream()
.flatMap(HoodieFileGroup::getAllFileSlices).sorted(new FileSliceComparator())
.collect(Collectors.toList());

allFileSlicesFromMeta = filterFileSliceBasedOnUnderDeletionFiles(fileSlicesFromMeta, baseFilesUnderDeletion);
allFileSlicesFromFS = filterFileSliceBasedOnUnderDeletionFiles(fileSlicesFromFS, baseFilesUnderDeletion);
} else {
allFileSlicesFromMeta = metadataTableBasedContext
.getSortedAllFileGroupList(partitionPath).stream()
.flatMap(HoodieFileGroup::getAllFileSlices).sorted(new FileSliceComparator())
.collect(Collectors.toList());
allFileSlicesFromFS = fsBasedContext
.getSortedAllFileGroupList(partitionPath).stream()
.flatMap(HoodieFileGroup::getAllFileSlices).sorted(new FileSliceComparator())
.collect(Collectors.toList());
}

LOG.debug("All file slices from metadata: " + allFileSlicesFromMeta + ". For partitions " + partitionPath);
LOG.debug("All file slices from direct listing: " + allFileSlicesFromFS + ". For partitions " + partitionPath);
Expand All @@ -459,10 +537,20 @@ private void validateAllFileGroups(
*/
private void validateLatestBaseFiles(
HoodieMetadataValidationContext metadataTableBasedContext,
HoodieMetadataValidationContext fsBasedContext, String partitionPath) {
HoodieMetadataValidationContext fsBasedContext,
String partitionPath,
List<String> baseFilesUnderDeletion) {

List<HoodieBaseFile> latestFilesFromMetadata;
List<HoodieBaseFile> latestFilesFromFS;

List<HoodieBaseFile> latestFilesFromMetadata = metadataTableBasedContext.getSortedLatestBaseFileList(partitionPath);
List<HoodieBaseFile> latestFilesFromFS = fsBasedContext.getSortedLatestBaseFileList(partitionPath);
if (!baseFilesUnderDeletion.isEmpty()) {
latestFilesFromMetadata = filterBaseFileBasedOnUnderDeletionFiles(metadataTableBasedContext.getSortedLatestBaseFileList(partitionPath), baseFilesUnderDeletion);
latestFilesFromFS = filterBaseFileBasedOnUnderDeletionFiles(fsBasedContext.getSortedLatestBaseFileList(partitionPath), baseFilesUnderDeletion);
} else {
latestFilesFromMetadata = metadataTableBasedContext.getSortedLatestBaseFileList(partitionPath);
latestFilesFromFS = fsBasedContext.getSortedLatestBaseFileList(partitionPath);
}

LOG.debug("Latest base file from metadata: " + latestFilesFromMetadata + ". For partitions " + partitionPath);
LOG.debug("Latest base file from direct listing: " + latestFilesFromFS + ". For partitions " + partitionPath);
Expand All @@ -483,10 +571,19 @@ private void validateLatestBaseFiles(
*/
private void validateLatestFileSlices(
HoodieMetadataValidationContext metadataTableBasedContext,
HoodieMetadataValidationContext fsBasedContext, String partitionPath) {

List<FileSlice> latestFileSlicesFromMetadataTable = metadataTableBasedContext.getSortedLatestFileSliceList(partitionPath);
List<FileSlice> latestFileSlicesFromFS = fsBasedContext.getSortedLatestFileSliceList(partitionPath);
HoodieMetadataValidationContext fsBasedContext,
String partitionPath,
List<String> baseFilesUnderDeletion) {
List<FileSlice> latestFileSlicesFromMetadataTable;
List<FileSlice> latestFileSlicesFromFS;

if (!baseFilesUnderDeletion.isEmpty()) {
latestFileSlicesFromMetadataTable = filterFileSliceBasedOnUnderDeletionFiles(metadataTableBasedContext.getSortedLatestFileSliceList(partitionPath), baseFilesUnderDeletion);
latestFileSlicesFromFS = filterFileSliceBasedOnUnderDeletionFiles(fsBasedContext.getSortedLatestFileSliceList(partitionPath), baseFilesUnderDeletion);
} else {
latestFileSlicesFromMetadataTable = metadataTableBasedContext.getSortedLatestFileSliceList(partitionPath);
latestFileSlicesFromFS = fsBasedContext.getSortedLatestFileSliceList(partitionPath);
}

LOG.debug("Latest file list from metadata: " + latestFileSlicesFromMetadataTable + ". For partition " + partitionPath);
LOG.debug("Latest file list from direct listing: " + latestFileSlicesFromFS + ". For partition " + partitionPath);
Expand All @@ -495,11 +592,31 @@ private void validateLatestFileSlices(
LOG.info("Validation of getLatestFileSlices succeeded for partition " + partitionPath);
}

private List<FileSlice> filterFileSliceBasedOnUnderDeletionFiles(List<FileSlice> sortedLatestFileSliceList, List<String> baseFilesUnderDeletion) {

@yihua yihua Mar 13, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could baseFilesUnderDeletion be a Set<String> to speed up lookup?

@zhangyue19921010 zhangyue19921010 Mar 14, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Make sense! changed. Thanks a lot for your reviewing!

return sortedLatestFileSliceList.stream()
.filter(fileSlice -> {
if (!fileSlice.getBaseFile().isPresent()) {
return true;
} else {
return !baseFilesUnderDeletion.contains(fileSlice.getBaseFile().get().getFileName());
}
}).collect(Collectors.toList());
}

private List<HoodieBaseFile> filterBaseFileBasedOnUnderDeletionFiles(List<HoodieBaseFile> sortedBaseFileList, List<String> baseFilesUnderDeletion) {
return sortedBaseFileList.stream()
.filter(baseFile -> {
return !baseFilesUnderDeletion.contains(baseFile.getFileName());
}).collect(Collectors.toList());
}

private void validateAllColumnStats(
HoodieMetadataValidationContext metadataTableBasedContext,
HoodieMetadataValidationContext fsBasedContext, String partitionPath) {
List<String> latestBaseFilenameList = fsBasedContext.getSortedLatestBaseFileList(partitionPath)
.stream().map(BaseFile::getFileName).collect(Collectors.toList());
HoodieMetadataValidationContext fsBasedContext,
String partitionPath,
List<String> baseFilesUnderDeletion) {

List<String> latestBaseFilenameList = getLatestBaseFileNames(fsBasedContext, partitionPath, baseFilesUnderDeletion);
List<HoodieColumnRangeMetadata<String>> metadataBasedColStats = metadataTableBasedContext
.getSortedColumnStatsList(partitionPath, latestBaseFilenameList);
List<HoodieColumnRangeMetadata<String>> fsBasedColStats = fsBasedContext
Expand All @@ -512,9 +629,11 @@ private void validateAllColumnStats(

private void validateBloomFilters(
HoodieMetadataValidationContext metadataTableBasedContext,
HoodieMetadataValidationContext fsBasedContext, String partitionPath) {
List<String> latestBaseFilenameList = fsBasedContext.getSortedLatestBaseFileList(partitionPath)
.stream().map(BaseFile::getFileName).collect(Collectors.toList());
HoodieMetadataValidationContext fsBasedContext,
String partitionPath,
List<String> baseFilesUnderDeletion) {

List<String> latestBaseFilenameList = getLatestBaseFileNames(fsBasedContext, partitionPath, baseFilesUnderDeletion);
List<BloomFilterData> metadataBasedBloomFilters = metadataTableBasedContext
.getSortedBloomFilterList(partitionPath, latestBaseFilenameList);
List<BloomFilterData> fsBasedBloomFilters = fsBasedContext
Expand All @@ -525,6 +644,19 @@ private void validateBloomFilters(
LOG.info("Validation of bloom filters succeeded for partition " + partitionPath);
}

private List<String> getLatestBaseFileNames(HoodieMetadataValidationContext fsBasedContext, String partitionPath, List<String> baseFilesUnderDeletion) {
List<String> latestBaseFilenameList;
if (!baseFilesUnderDeletion.isEmpty()) {
List<HoodieBaseFile> sortedLatestBaseFileList = fsBasedContext.getSortedLatestBaseFileList(partitionPath);
latestBaseFilenameList = filterBaseFileBasedOnUnderDeletionFiles(sortedLatestBaseFileList, baseFilesUnderDeletion)
.stream().map(BaseFile::getFileName).collect(Collectors.toList());
} else {
latestBaseFilenameList = fsBasedContext.getSortedLatestBaseFileList(partitionPath)
.stream().map(BaseFile::getFileName).collect(Collectors.toList());
}
return latestBaseFilenameList;
}

private <T> void validate(
List<T> infoListFromMetadataTable, List<T> infoListFromFS, String partitionPath, String label) {
if (infoListFromMetadataTable.size() != infoListFromFS.size()
Expand Down