This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 43a617f86cb [HUDI-6689] Add record index validation in MDT validator
(#9437)
43a617f86cb is described below
commit 43a617f86cb6daa2407873ca9e6b13320a4f8e96
Author: Y Ethan Guo <[email protected]>
AuthorDate: Tue Aug 15 09:40:43 2023 -0700
[HUDI-6689] Add record index validation in MDT validator (#9437)
This PR adds the validation of record index in MDT validator
(`HoodieMetadataTableValidator`). The following validation modes are added:
- Record index count validation (with CLI config
`--validate-record-index-count`): validate the number of entries in the record
index, which should be equal to the number of record keys in the latest
snapshot of the table.
- Record index content validation (with CLI config
`--validate-record-index-content`): validate the content of the record index so
that each record key should have the correct location, and there is no
additional or missing entry. Two more configs are added for this mode: (1)
`--num-record-index-error-samples`: number of error samples to show for record
index validation when there are mismatches, (2) `--record-index-parallelism`:
parallelism for joining record index entries with data [...]
---
.../hudi/metadata/HoodieMetadataPayload.java | 19 +-
.../hudi/metadata/HoodieTableMetadataUtil.java | 71 +++++-
.../utilities/HoodieMetadataTableValidator.java | 272 +++++++++++++++++++--
3 files changed, 319 insertions(+), 43 deletions(-)
diff --git
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataPayload.java
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataPayload.java
index 8d5114a76bc..04ffc98e840 100644
---
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataPayload.java
+++
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataPayload.java
@@ -158,7 +158,7 @@ public class HoodieMetadataPayload implements
HoodieRecordPayload<HoodieMetadata
/**
* FileIndex value saved in record index record when the fileId has no index
(old format of base filename)
*/
- private static final int RECORD_INDEX_MISSING_FILEINDEX_FALLBACK = -1;
+ public static final int RECORD_INDEX_MISSING_FILEINDEX_FALLBACK = -1;
/**
* NOTE: PLEASE READ CAREFULLY
@@ -761,22 +761,7 @@ public class HoodieMetadataPayload implements
HoodieRecordPayload<HoodieMetadata
* If this is a record-level index entry, returns the file to which this is
mapped.
*/
public HoodieRecordGlobalLocation getRecordGlobalLocation() {
- final String partition = recordIndexMetadata.getPartitionName();
- String fileId = null;
- if (recordIndexMetadata.getFileIdEncoding() == 0) {
- // encoding 0 refers to UUID based fileID
- final UUID uuid = new UUID(recordIndexMetadata.getFileIdHighBits(),
recordIndexMetadata.getFileIdLowBits());
- fileId = uuid.toString();
- if (recordIndexMetadata.getFileIndex() !=
RECORD_INDEX_MISSING_FILEINDEX_FALLBACK) {
- fileId += "-" + recordIndexMetadata.getFileIndex();
- }
- } else {
- // encoding 1 refers to no encoding. fileID as is.
- fileId = recordIndexMetadata.getFileId();
- }
-
- final java.util.Date instantDate = new
java.util.Date(recordIndexMetadata.getInstantTime());
- return new HoodieRecordGlobalLocation(partition,
HoodieActiveTimeline.formatDate(instantDate), fileId);
+ return
HoodieTableMetadataUtil.getLocationFromRecordIndexInfo(recordIndexMetadata);
}
public boolean isDeleted() {
diff --git
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
index 08fc663fbad..57f6b405628 100644
---
a/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
+++
b/hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java
@@ -18,10 +18,10 @@
package org.apache.hudi.metadata;
-import org.apache.hadoop.conf.Configuration;
import org.apache.hudi.avro.ConvertingGenericData;
import org.apache.hudi.avro.model.HoodieCleanMetadata;
import org.apache.hudi.avro.model.HoodieMetadataColumnStats;
+import org.apache.hudi.avro.model.HoodieRecordIndexInfo;
import org.apache.hudi.avro.model.HoodieRestoreMetadata;
import org.apache.hudi.avro.model.HoodieRollbackMetadata;
import org.apache.hudi.avro.model.HoodieRollbackPlan;
@@ -39,6 +39,7 @@ import org.apache.hudi.common.model.HoodieDeltaWriteStat;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
+import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
@@ -66,6 +67,7 @@ import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
@@ -89,6 +91,7 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.UUID;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collector;
@@ -105,6 +108,7 @@ import static
org.apache.hudi.avro.HoodieAvroUtils.unwrapAvroValueWrapper;
import static
org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator.MILLIS_INSTANT_ID_LENGTH;
import static org.apache.hudi.common.util.StringUtils.isNullOrEmpty;
import static org.apache.hudi.common.util.ValidationUtils.checkState;
+import static
org.apache.hudi.metadata.HoodieMetadataPayload.RECORD_INDEX_MISSING_FILEINDEX_FALLBACK;
import static
org.apache.hudi.metadata.HoodieTableMetadata.EMPTY_PARTITION_NAME;
import static
org.apache.hudi.metadata.HoodieTableMetadata.NON_PARTITIONED_NAME;
import static
org.apache.hudi.metadata.HoodieTableMetadata.SOLO_COMMIT_TIMESTAMP;
@@ -529,8 +533,8 @@ public class HoodieTableMetadataUtil {
}
public static Map<MetadataPartitionType, HoodieData<HoodieRecord>>
convertMissingPartitionRecords(HoodieEngineContext engineContext,
-
List<String> deletedPartitions, Map<String, Map<String, Long>> filesAdded,
-
Map<String, List<String>> filesDeleted, String instantTime) {
+
List<String> deletedPartitions, Map<String, Map<String,
Long>> filesAdded,
+
Map<String, List<String>> filesDeleted, String
instantTime) {
List<HoodieRecord> records = new LinkedList<>();
int[] fileDeleteCount = {0};
int[] filesAddedCount = {0};
@@ -1069,8 +1073,8 @@ public class HoodieTableMetadataUtil {
}
private static Stream<HoodieRecord>
translateWriteStatToColumnStats(HoodieWriteStat writeStat,
-
HoodieTableMetaClient datasetMetaClient,
-
List<String> columnsToIndex) {
+
HoodieTableMetaClient datasetMetaClient,
+
List<String> columnsToIndex) {
if (writeStat instanceof HoodieDeltaWriteStat && ((HoodieDeltaWriteStat)
writeStat).getColumnStats().isPresent()) {
Map<String, HoodieColumnRangeMetadata<Comparable>> columnRangeMap =
((HoodieDeltaWriteStat) writeStat).getColumnStats().get();
Collection<HoodieColumnRangeMetadata<Comparable>>
columnRangeMetadataList = columnRangeMap.values();
@@ -1332,7 +1336,7 @@ public class HoodieTableMetadataUtil {
*/
public static boolean isIndexingCommit(String instantTime) {
return instantTime.length() == MILLIS_INSTANT_ID_LENGTH +
OperationSuffix.METADATA_INDEXER.getSuffix().length()
- &&
instantTime.endsWith(OperationSuffix.METADATA_INDEXER.getSuffix());
+ && instantTime.endsWith(OperationSuffix.METADATA_INDEXER.getSuffix());
}
/**
@@ -1457,7 +1461,7 @@ public class HoodieTableMetadataUtil {
if (backup) {
final Path metadataPartitionBackupPath = new
Path(metadataTablePartitionPath.getParent().getParent(),
- String.format(".metadata_%s_%s",
partitionType.getPartitionPath(), HoodieActiveTimeline.createNewInstantTime()));
+ String.format(".metadata_%s_%s", partitionType.getPartitionPath(),
HoodieActiveTimeline.createNewInstantTime()));
LOG.info(String.format("Backing up MDT partition %s to %s before
deletion", partitionType, metadataPartitionBackupPath));
try {
if (fs.rename(metadataTablePartitionPath,
metadataPartitionBackupPath)) {
@@ -1586,7 +1590,7 @@ public class HoodieTableMetadataUtil {
* @return The estimated number of file groups.
*/
public static int estimateFileGroupCount(MetadataPartitionType
partitionType, long recordCount, int averageRecordSize, int minFileGroupCount,
- int maxFileGroupCount, float growthFactor, int maxFileGroupSizeBytes) {
+ int maxFileGroupCount, float
growthFactor, int maxFileGroupSizeBytes) {
int fileGroupCount;
// If a fixed number of file groups are desired
@@ -1640,4 +1644,55 @@ public class HoodieTableMetadataUtil {
}
return false;
}
+
+ /**
+ * Gets the location from record index content.
+ *
+ * @param recordIndexInfo {@link HoodieRecordIndexInfo} instance.
+ * @return {@link HoodieRecordGlobalLocation} containing the location.
+ */
+ public static HoodieRecordGlobalLocation
getLocationFromRecordIndexInfo(HoodieRecordIndexInfo recordIndexInfo) {
+ return getLocationFromRecordIndexInfo(
+ recordIndexInfo.getPartitionName(),
recordIndexInfo.getFileIdEncoding(),
+ recordIndexInfo.getFileIdHighBits(),
recordIndexInfo.getFileIdLowBits(),
+ recordIndexInfo.getFileIndex(), recordIndexInfo.getFileId(),
+ recordIndexInfo.getInstantTime());
+ }
+
+ /**
+ * Gets the location from record index content.
+ * Note that, a UUID based fileId is stored as 3 pieces in record index
(fileIdHighBits,
+ * fileIdLowBits and fileIndex). FileID format is {UUID}-{fileIndex}.
+ * The arguments are consistent with what {@link HoodieRecordIndexInfo}
contains.
+ *
+ * @param partition The partition name the record belongs to.
+ * @param fileIdEncoding FileId encoding. Possible values are 0 and 1. O
represents UUID based
+ * fileID, and 1 represents raw string format of the
fileId.
+ * @param fileIdHighBits High 64 bits if the fileId is based on UUID format.
+ * @param fileIdLowBits Low 64 bits if the fileId is based on UUID format.
+ * @param fileIndex Index representing file index which is used to
re-construct UUID based fileID.
+ * @param originalFileId FileId of the location where record belongs to.
+ * When the encoding is 1, fileID is stored in raw
string format.
+ * @param instantTime Epoch time in millisecond representing the commit
time at which record was added.
+ * @return {@link HoodieRecordGlobalLocation} containing the location.
+ */
+ public static HoodieRecordGlobalLocation getLocationFromRecordIndexInfo(
+ String partition, int fileIdEncoding, long fileIdHighBits, long
fileIdLowBits,
+ int fileIndex, String originalFileId, Long instantTime) {
+ String fileId = null;
+ if (fileIdEncoding == 0) {
+ // encoding 0 refers to UUID based fileID
+ final UUID uuid = new UUID(fileIdHighBits, fileIdLowBits);
+ fileId = uuid.toString();
+ if (fileIndex != RECORD_INDEX_MISSING_FILEINDEX_FALLBACK) {
+ fileId += "-" + fileIndex;
+ }
+ } else {
+ // encoding 1 refers to no encoding. fileID as is.
+ fileId = originalFileId;
+ }
+
+ final java.util.Date instantDate = new java.util.Date(instantTime);
+ return new HoodieRecordGlobalLocation(partition,
HoodieActiveTimeline.formatDate(instantDate), fileId);
+ }
}
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
index 29e59df6935..45c12fcfe28 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieMetadataTableValidator.java
@@ -36,6 +36,7 @@ import org.apache.hudi.common.model.HoodieFileGroup;
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodiePartitionMetadata;
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
+import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.TableSchemaResolver;
@@ -68,7 +69,10 @@ import org.apache.hadoop.fs.Path;
import org.apache.parquet.avro.AvroSchemaConverter;
import org.apache.parquet.schema.MessageType;
import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.java.Optional;
+import org.apache.spark.sql.functions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -89,20 +93,30 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
+import scala.Tuple2;
+
+import static
org.apache.hudi.common.model.HoodieRecord.FILENAME_METADATA_FIELD;
+import static
org.apache.hudi.common.model.HoodieRecord.PARTITION_PATH_METADATA_FIELD;
+import static
org.apache.hudi.common.model.HoodieRecord.RECORD_KEY_METADATA_FIELD;
import static
org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType.INSTANT_TIME;
import static
org.apache.hudi.common.table.timeline.HoodieTimeline.LESSER_THAN_OR_EQUALS;
import static
org.apache.hudi.hadoop.CachingPath.getPathWithoutSchemeAndAuthority;
+import static
org.apache.hudi.metadata.HoodieTableMetadata.getMetadataTableBasePath;
/**
* A validator with spark-submit to compare information, such as partitions,
file listing, index, etc.,
* between metadata table and filesystem.
* <p>
- * There are five validation tasks, that can be enabled independently through
the following CLI options:
+ * There are seven validation tasks, that can be enabled independently through
the following CLI options:
* - `--validate-latest-file-slices`: validate the latest file slices for all
partitions.
* - `--validate-latest-base-files`: validate the latest base files for all
partitions.
* - `--validate-all-file-groups`: validate all file groups, and all file
slices within file groups.
* - `--validate-all-column-stats`: validate column stats for all columns in
the schema
* - `--validate-bloom-filters`: validate bloom filters of base files
+ * - `--validate-record-index-count`: validate the number of entries in the
record index, which
+ * should be equal to the number of record keys in the latest snapshot of the
table.
+ * - `--validate-record-index-content`: validate the content of the record
index so that each
+ * record key should have the correct location, and there is no additional or
missing entry.
* <p>
* If the Hudi table is on the local file system, the base path passed to
`--base-path` must have
* "file:" prefix to avoid validation failure.
@@ -194,6 +208,12 @@ public class HoodieMetadataTableValidator implements
Serializable {
if (cfg.validateBloomFilters) {
labelList.add("validate-bloom-filters");
}
+ if (cfg.validateRecordIndexCount) {
+ labelList.add("validate-record-index-count");
+ }
+ if (cfg.validateRecordIndexContent) {
+ labelList.add("validate-record-index-content");
+ }
return String.join(",", labelList);
}
@@ -235,6 +255,23 @@ public class HoodieMetadataTableValidator implements
Serializable {
@Parameter(names = {"--validate-bloom-filters"}, description = "Validate
bloom filters of base files", required = false)
public boolean validateBloomFilters = false;
+ @Parameter(names = {"--validate-record-index-count"},
+ description = "Validate the number of entries in the record index,
which should be equal "
+ + "to the number of record keys in the latest snapshot of the
table",
+ required = false)
+ public boolean validateRecordIndexCount = false;
+
+ @Parameter(names = {"--validate-record-index-content"},
+ description = "Validate the content of the record index so that each
record key should "
+ + "have the correct location, and there is no additional or
missing entry",
+ required = false)
+ public boolean validateRecordIndexContent = false;
+
+ @Parameter(names = {"--num-record-index-error-samples"},
+ description = "Number of error samples to show for record index
validation",
+ required = false)
+ public int numRecordIndexErrorSamples = 100;
+
@Parameter(names = {"--min-validate-interval-seconds"},
description = "the min validate interval of each validate when set
--continuous, default is 10 minutes.")
public Integer minValidateIntervalSeconds = 10 * 60;
@@ -242,6 +279,9 @@ public class HoodieMetadataTableValidator implements
Serializable {
@Parameter(names = {"--parallelism", "-pl"}, description = "Parallelism
for valuation", required = false)
public int parallelism = 200;
+ @Parameter(names = {"--record-index-parallelism", "-rpl"}, description =
"Parallelism for validating record index", required = false)
+ public int recordIndexParallelism = 100;
+
@Parameter(names = {"--ignore-failed", "-ig"}, description = "Ignore
metadata validate failure and continue.", required = false)
public boolean ignoreFailed = false;
@@ -276,11 +316,15 @@ public class HoodieMetadataTableValidator implements
Serializable {
+ " --validate-all-file-groups " + validateAllFileGroups + ", \n"
+ " --validate-all-column-stats " + validateAllColumnStats + ", \n"
+ " --validate-bloom-filters " + validateBloomFilters + ", \n"
+ + " --validate-record-index-count " + validateRecordIndexCount +
", \n"
+ + " --validate-record-index-content " + validateRecordIndexContent
+ ", \n"
+ + " --num-record-index-error-samples " +
numRecordIndexErrorSamples + ", \n"
+ " --continuous " + continuous + ", \n"
+ " --skip-data-files-for-cleaning " + skipDataFilesForCleaning +
", \n"
+ " --ignore-failed " + ignoreFailed + ", \n"
+ " --min-validate-interval-seconds " + minValidateIntervalSeconds
+ ", \n"
+ " --parallelism " + parallelism + ", \n"
+ + " --record-index-parallelism " + recordIndexParallelism + ", \n"
+ " --spark-master " + sparkMaster + ", \n"
+ " --spark-memory " + sparkMemory + ", \n"
+ " --assumeDatePartitioning-memory " + assumeDatePartitioning +
", \n"
@@ -306,8 +350,12 @@ public class HoodieMetadataTableValidator implements
Serializable {
&& Objects.equals(validateAllFileGroups,
config.validateAllFileGroups)
&& Objects.equals(validateAllColumnStats,
config.validateAllColumnStats)
&& Objects.equals(validateBloomFilters, config.validateBloomFilters)
+ && Objects.equals(validateRecordIndexCount,
config.validateRecordIndexCount)
+ && Objects.equals(validateRecordIndexContent,
config.validateRecordIndexContent)
+ && Objects.equals(numRecordIndexErrorSamples,
config.numRecordIndexErrorSamples)
&& Objects.equals(minValidateIntervalSeconds,
config.minValidateIntervalSeconds)
&& Objects.equals(parallelism, config.parallelism)
+ && Objects.equals(recordIndexParallelism,
config.recordIndexParallelism)
&& Objects.equals(ignoreFailed, config.ignoreFailed)
&& Objects.equals(sparkMaster, config.sparkMaster)
&& Objects.equals(sparkMemory, config.sparkMemory)
@@ -318,9 +366,11 @@ public class HoodieMetadataTableValidator implements
Serializable {
@Override
public int hashCode() {
- return Objects.hash(basePath, continuous, skipDataFilesForCleaning,
validateLatestFileSlices, validateLatestBaseFiles,
- validateAllFileGroups, validateAllColumnStats, validateBloomFilters,
minValidateIntervalSeconds,
- parallelism, ignoreFailed, sparkMaster, sparkMemory,
assumeDatePartitioning, propsFilePath, configs, help);
+ return Objects.hash(basePath, continuous, skipDataFilesForCleaning,
validateLatestFileSlices,
+ validateLatestBaseFiles, validateAllFileGroups,
validateAllColumnStats, validateBloomFilters,
+ validateRecordIndexCount, validateRecordIndexContent,
numRecordIndexErrorSamples,
+ minValidateIntervalSeconds, parallelism, recordIndexParallelism,
ignoreFailed,
+ sparkMaster, sparkMemory, assumeDatePartitioning, propsFilePath,
configs, help);
}
}
@@ -444,21 +494,34 @@ public class HoodieMetadataTableValidator implements
Serializable {
HoodieMetadataValidationContext fsBasedContext =
new HoodieMetadataValidationContext(engineContext, cfg,
metaClient, false)) {
Set<String> finalBaseFilesForCleaning = baseFilesForCleaning;
- List<Pair<Boolean, String>> result =
engineContext.parallelize(allPartitions,
allPartitions.size()).map(partitionPath -> {
- try {
- validateFilesInPartition(metadataTableBasedContext, fsBasedContext,
partitionPath, finalBaseFilesForCleaning);
- LOG.info(String.format("Metadata table validation succeeded for
partition %s (partition %s)", partitionPath, taskLabels));
- return Pair.of(true, "");
- } catch (HoodieValidationException e) {
- LOG.error(
- String.format("Metadata table validation failed for partition %s
due to HoodieValidationException (partition %s)",
- partitionPath, taskLabels), e);
- if (!cfg.ignoreFailed) {
- throw e;
- }
- return Pair.of(false, e.getMessage() + " for partition: " +
partitionPath);
+ List<Pair<Boolean, String>> result = new ArrayList<>(
+ engineContext.parallelize(allPartitions,
allPartitions.size()).map(partitionPath -> {
+ try {
+ validateFilesInPartition(metadataTableBasedContext,
fsBasedContext, partitionPath, finalBaseFilesForCleaning);
+ LOG.info(String.format("Metadata table validation succeeded for
partition %s (partition %s)", partitionPath, taskLabels));
+ return Pair.of(true, "");
+ } catch (HoodieValidationException e) {
+ LOG.error(
+ String.format("Metadata table validation failed for
partition %s due to HoodieValidationException (partition %s)",
+ partitionPath, taskLabels), e);
+ if (!cfg.ignoreFailed) {
+ throw e;
+ }
+ return Pair.of(false, e.getMessage() + " for partition: " +
partitionPath);
+ }
+ }).collectAsList());
+
+ try {
+ validateRecordIndex(engineContext, metaClient,
metadataTableBasedContext.getTableMetadata());
+ result.add(Pair.of(true, ""));
+ } catch (HoodieValidationException e) {
+ LOG.error(
+ "Metadata table validation failed due to HoodieValidationException
in record index validation", e);
+ if (!cfg.ignoreFailed) {
+ throw e;
}
- }).collectAsList();
+ result.add(Pair.of(false, e.getMessage()));
+ }
for (Pair<Boolean, String> res : result) {
finalResult &= res.getKey();
@@ -741,6 +804,174 @@ public class HoodieMetadataTableValidator implements
Serializable {
validate(metadataBasedBloomFilters, fsBasedBloomFilters, partitionPath,
"bloom filters");
}
+ private void validateRecordIndex(HoodieSparkEngineContext sparkEngineContext,
+ HoodieTableMetaClient metaClient,
+ HoodieTableMetadata tableMetadata) {
+ if (cfg.validateRecordIndexContent) {
+ validateRecordIndexContent(sparkEngineContext, metaClient,
tableMetadata);
+ } else if (cfg.validateRecordIndexCount) {
+ validateRecordIndexCount(sparkEngineContext, metaClient);
+ }
+ }
+
+ private void validateRecordIndexCount(HoodieSparkEngineContext
sparkEngineContext,
+ HoodieTableMetaClient metaClient) {
+ String basePath = metaClient.getBasePathV2().toString();
+ long countKeyFromTable =
sparkEngineContext.getSqlContext().read().format("hudi")
+ .load(basePath)
+ .select(RECORD_KEY_METADATA_FIELD)
+ .count();
+ long countKeyFromRecordIndex =
sparkEngineContext.getSqlContext().read().format("hudi")
+ .load(getMetadataTableBasePath(basePath))
+ .select("key")
+ .filter("type = 5")
+ .count();
+
+ if (countKeyFromTable != countKeyFromRecordIndex) {
+ String message = String.format("Validation of record index count failed:
"
+ + "%s entries from record index metadata, %s keys from the data
table.",
+ countKeyFromRecordIndex, countKeyFromTable);
+ LOG.error(message);
+ throw new HoodieValidationException(message);
+ } else {
+ LOG.info(String.format(
+ "Validation of record index count succeeded: %s entries.",
countKeyFromRecordIndex));
+ }
+ }
+
+ private void validateRecordIndexContent(HoodieSparkEngineContext
sparkEngineContext,
+ HoodieTableMetaClient metaClient,
+ HoodieTableMetadata tableMetadata) {
+ String basePath = metaClient.getBasePathV2().toString();
+ JavaPairRDD<String, Pair<String, String>> keyToLocationOnFsRdd =
+ sparkEngineContext.getSqlContext().read().format("hudi").load(basePath)
+ .select(RECORD_KEY_METADATA_FIELD, PARTITION_PATH_METADATA_FIELD,
FILENAME_METADATA_FIELD)
+ .toJavaRDD()
+ .mapToPair(row -> new
Tuple2<>(row.getString(row.fieldIndex(RECORD_KEY_METADATA_FIELD)),
+
Pair.of(row.getString(row.fieldIndex(PARTITION_PATH_METADATA_FIELD)),
+
FSUtils.getFileId(row.getString(row.fieldIndex(FILENAME_METADATA_FIELD))))))
+ .cache();
+
+ JavaPairRDD<String, Pair<String, String>> keyToLocationFromRecordIndexRdd =
+ sparkEngineContext.getSqlContext().read().format("hudi")
+ .load(getMetadataTableBasePath(basePath))
+ .filter("type = 5")
+ .select(functions.col("key"),
+
functions.col("recordIndexMetadata.partitionName").as("partitionName"),
+
functions.col("recordIndexMetadata.fileIdHighBits").as("fileIdHighBits"),
+
functions.col("recordIndexMetadata.fileIdLowBits").as("fileIdLowBits"),
+ functions.col("recordIndexMetadata.fileIndex").as("fileIndex"),
+ functions.col("recordIndexMetadata.fileId").as("fileId"),
+
functions.col("recordIndexMetadata.instantTime").as("instantTime"),
+
functions.col("recordIndexMetadata.fileIdEncoding").as("fileIdEncoding"))
+ .toJavaRDD()
+ .mapToPair(row -> {
+ HoodieRecordGlobalLocation location =
HoodieTableMetadataUtil.getLocationFromRecordIndexInfo(
+ row.getString(row.fieldIndex("partitionName")),
+ row.getInt(row.fieldIndex("fileIdEncoding")),
+ row.getLong(row.fieldIndex("fileIdHighBits")),
+ row.getLong(row.fieldIndex("fileIdLowBits")),
+ row.getInt(row.fieldIndex("fileIndex")),
+ row.getString(row.fieldIndex("fileId")),
+ row.getLong(row.fieldIndex("instantTime")));
+ return new Tuple2<>(row.getString(row.fieldIndex("key")),
+ Pair.of(location.getPartitionPath(), location.getFileId()));
+ });
+
+ int numErrorSamples = cfg.numRecordIndexErrorSamples;
+ Pair<Long, List<String>> result =
keyToLocationOnFsRdd.fullOuterJoin(keyToLocationFromRecordIndexRdd,
cfg.recordIndexParallelism)
+ .map(e -> {
+ Optional<Pair<String, String>> locationOnFs = e._2._1;
+ Optional<Pair<String, String>> locationFromRecordIndex = e._2._2;
+ StringBuilder sb = new StringBuilder();
+ List<String> errorSampleList = new ArrayList<>();
+ if (locationOnFs.isPresent() && locationFromRecordIndex.isPresent())
{
+ if
(locationOnFs.get().getLeft().equals(locationFromRecordIndex.get().getLeft())
+ &&
locationOnFs.get().getRight().equals(locationFromRecordIndex.get().getRight()))
{
+ return Pair.of(0L, errorSampleList);
+ }
+ errorSampleList.add(constructLocationInfoString(locationOnFs,
locationFromRecordIndex));
+ return Pair.of(1L, errorSampleList);
+ }
+ if (!locationOnFs.isPresent() &&
!locationFromRecordIndex.isPresent()) {
+ return Pair.of(0L, errorSampleList);
+ }
+ errorSampleList.add(constructLocationInfoString(locationOnFs,
locationFromRecordIndex));
+ return Pair.of(1L, errorSampleList);
+ })
+ .reduce((pair1, pair2) -> {
+ long errorCount = pair1.getLeft() + pair2.getLeft();
+ List<String> list1 = pair1.getRight();
+ List<String> list2 = pair2.getRight();
+ if (!list1.isEmpty() && !list2.isEmpty()) {
+ if (list1.size() >= numErrorSamples) {
+ return Pair.of(errorCount, list1);
+ }
+ if (list2.size() >= numErrorSamples) {
+ return Pair.of(errorCount, list2);
+ }
+
+ List<String> resultList = new ArrayList<>();
+ if (list1.size() > list2.size()) {
+ resultList.addAll(list1);
+ for (String item : list2) {
+ resultList.add(item);
+ if (resultList.size() >= numErrorSamples) {
+ break;
+ }
+ }
+ } else {
+ resultList.addAll(list2);
+ for (String item : list1) {
+ resultList.add(item);
+ if (resultList.size() >= numErrorSamples) {
+ break;
+ }
+ }
+ }
+ return Pair.of(errorCount, resultList);
+ } else if (!list1.isEmpty()) {
+ return Pair.of(errorCount, list1);
+ } else {
+ return Pair.of(errorCount, list2);
+ }
+ });
+
+ long countKey = keyToLocationOnFsRdd.count();
+ keyToLocationOnFsRdd.unpersist();
+
+ long diffCount = result.getLeft();
+ if (diffCount > 0) {
+ String message = String.format("Validation of record index content
failed: "
+ + "%s keys (total %s) from the data table have wrong location in
record index "
+ + "metadata. Sample mismatches: %s",
+ diffCount, countKey, String.join(";", result.getRight()));
+ LOG.error(message);
+ throw new HoodieValidationException(message);
+ } else {
+ LOG.info(String.format(
+ "Validation of record index content succeeded: %s entries.",
countKey));
+ }
+ }
+
+ private String constructLocationInfoString(Optional<Pair<String, String>>
locationOnFs,
+ Optional<Pair<String, String>>
locationFromRecordIndex) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("FS: ");
+ if (locationOnFs.isPresent()) {
+ sb.append(locationOnFs.get());
+ } else {
+ sb.append("<empty>");
+ }
+ sb.append(", Record Index: ");
+ if (locationFromRecordIndex.isPresent()) {
+ sb.append(locationFromRecordIndex.get());
+ } else {
+ sb.append("<empty>");
+ }
+ return sb.toString();
+ }
+
private List<String> getLatestBaseFileNames(HoodieMetadataValidationContext
fsBasedContext, String partitionPath, Set<String> baseDataFilesForCleaning) {
List<String> latestBaseFilenameList;
if (!baseDataFilesForCleaning.isEmpty()) {
@@ -1050,6 +1281,7 @@ public class HoodieMetadataTableValidator implements
Serializable {
.enable(enableMetadataTable)
.withMetadataIndexBloomFilter(enableMetadataTable)
.withMetadataIndexColumnStats(enableMetadataTable)
+ .withEnableRecordIndex(enableMetadataTable)
.withAssumeDatePartitioning(cfg.assumeDatePartitioning)
.build();
this.fileSystemView =
FileSystemViewManager.createInMemoryFileSystemView(engineContext,
@@ -1064,6 +1296,10 @@ public class HoodieMetadataTableValidator implements
Serializable {
return metaClient;
}
+ public HoodieTableMetadata getTableMetadata() {
+ return tableMetadata;
+ }
+
public List<HoodieBaseFile> getSortedLatestBaseFileList(String
partitionPath) {
return fileSystemView.getLatestBaseFiles(partitionPath)
.sorted(new HoodieBaseFileComparator()).collect(Collectors.toList());
