Skip to content

fix(utilities): include metadata table index-init instants in the record index validation snapshot #19395

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

Open
Davis-Zhang-Onehouse wants to merge 1 commit into
base: master
Choose a base branch
from
+162 −2
Open

fix(utilities): include metadata table index-init instants in the record index validation snapshot #19395

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
42 changes: 40 additions & 2 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 @@ -61,6 +61,7 @@
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.timeline.InstantComparison;
import org.apache.hudi.common.table.timeline.TimelineUtils;
import org.apache.hudi.common.table.view.FileSystemViewManager;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import org.apache.hudi.common.table.view.FileSystemViewStorageType;
Expand Down Expand Up @@ -111,11 +112,13 @@
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
Expand Down Expand Up @@ -208,6 +211,10 @@ public class HoodieMetadataTableValidator implements Serializable {

private static final long serialVersionUID = 1L;

// Advance the metadata table query instant by this much so that instants derived from a data table
// instant, which carry a three-digit suffix, fall inside the queried window. See #metadataTableInstantFor.
private static final long METADATA_INSTANT_LOOKAHEAD_MS = 1;

// Spark context
private transient JavaSparkContext jsc;
// config
Expand Down Expand Up @@ -1235,7 +1242,7 @@ private void validateRecordIndexCount(HoodieSparkEngineContext sparkEngineContex
.select(RECORD_KEY_METADATA_FIELD)
.count();
long countKeyFromRecordIndex = sparkEngineContext.getSqlContext().read().format("hudi")
.option(DataSourceReadOptions.TIME_TRAVEL_AS_OF_INSTANT().key(),latestCompletedCommit)
.option(DataSourceReadOptions.TIME_TRAVEL_AS_OF_INSTANT().key(), metadataTableInstantFor(latestCompletedCommit))
.load(getMetadataTableBasePath(basePath))
.select("key")
.filter("type = 5")
Expand Down Expand Up @@ -1337,6 +1344,37 @@ private void validateRecordIndexContent(HoodieSparkEngineContext sparkEngineCont
}
}

/**
* Returns the instant to query the metadata table with, so that the snapshot reflects the data
* table as of {@code dataTableInstant}.
* <p>
* Metadata table instants derived from a data table instant carry a three-digit numeric suffix:
* partition initialization appends 010 and up (see
* {@code HoodieTableMetadataUtil#createIndexInitTimestamp}), and metadata-table-internal
* compaction, clean, restore, indexing, log compaction and rollback append 001 to 006. Hudi
* compares instants as strings, so every one of those derived instants sorts AFTER the bare data
* instant, and a snapshot taken as of the data instant itself excludes them - leaving, for
* instance, the record index unreadable until the data table receives another commit.
* <p>
* The bound is therefore advanced by a single millisecond. That is strictly greater than any
* {@code <dataTableInstant><suffix>} (which shares the whole 17-character prefix and so compares
* lower), while still being a valid {@code yyyyMMddHHmmssSSS} instant - the time travel option
* rejects anything else, see {@code HoodieSqlCommonUtils#formatQueryInstant}. Instants that are
* not timestamps (legacy or test instants such as "100") are returned unchanged; they have no
* metadata table counterpart to include.
*/
@VisibleForTesting
static String metadataTableInstantFor(String dataTableInstant) {
try {
Date dataTableInstantDate = TimelineUtils.parseDateFromInstantTime(dataTableInstant);
return TimelineUtils.formatDate(new Date(dataTableInstantDate.getTime() + METADATA_INSTANT_LOOKAHEAD_MS));
} catch (ParseException e) {
log.warn("Cannot parse instant {} as a timestamp; querying the metadata table as of it verbatim",
dataTableInstant);
return dataTableInstant;
}
}

@VisibleForTesting
JavaPairRDD<String, Pair<String, String>> getRecordLocationsFromFSBasedListing(HoodieSparkEngineContext sparkEngineContext,
String basePath,
Expand All @@ -1357,7 +1395,7 @@ JavaPairRDD<String, Pair<String, String>> getRecordLocationsFromRLI(HoodieSparkE
String basePath,
String latestCompletedCommit) {
return sparkEngineContext.getSqlContext().read().format("hudi")
.option(DataSourceReadOptions.TIME_TRAVEL_AS_OF_INSTANT().key(), latestCompletedCommit)
.option(DataSourceReadOptions.TIME_TRAVEL_AS_OF_INSTANT().key(), metadataTableInstantFor(latestCompletedCommit))
.load(getMetadataTableBasePath(basePath))
.filter("type = 5")
.select(functions.col("key"),
Expand Down
122 changes: 122 additions & 0 deletions hudi-utilities/src/test/java/org/apache/hudi/utilities/TestHoodieMetadataTableValidator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableVersion;
import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.HoodieLogFormatWriter;
import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
Expand All @@ -45,6 +46,7 @@
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.timeline.InstantComparison;
import org.apache.hudi.common.table.timeline.TimeGenerator;
import org.apache.hudi.common.table.timeline.TimeGenerators;
import org.apache.hudi.common.table.timeline.TimelineUtils;
Expand All @@ -57,9 +59,14 @@
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieCompactionConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieValidationException;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.metadata.HoodieTableMetadataWriter;
import org.apache.hudi.metadata.MetadataPartitionType;
import org.apache.hudi.metadata.SparkMetadataWriterFactory;
import org.apache.hudi.metadata.stats.HoodieColumnRangeMetadata;
import org.apache.hudi.metadata.stats.ValueMetadata;
import org.apache.hudi.storage.HoodieStorage;
Expand Down Expand Up @@ -1678,4 +1685,119 @@ private void mockPartitionWithFiles(List<String> partition1, HoodieStorage stora
when(storage.listFiles(new StoragePath(basePath + "/" + partition))).thenReturn(Collections.singletonList(storagePathInfo));
}
}

/**
* On a table version 6 metadata table the partition-initialization deltacommits are the data instant
* they were derived from with a three-digit suffix appended (010 for FILES, 011 for RECORD_INDEX) -
* see {@code HoodieBackedTableMetadataWriterTableVersionSix#createIndexInitTimestamp}. Those instants
* sort AFTER the bare data instant under Hudi's lexicographic instant comparison, so a metadata-table
* snapshot taken as of the data table's latest completed commit must still include them. When the data
* table has no commit after the metadata table was initialized, failing to do so makes the record index
* read back empty and every data-table key is reported as missing from it.
* <p>
* Table version 8 and above are unaffected: {@code HoodieBackedTableMetadataWriter#generateUniqueInstantTime}
* derives the init instants from {@code SOLO_COMMIT_TIMESTAMP}, which sorts below every data instant.
*/
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testRecordIndexValidationWhenMdtInitializedAtLatestDataCommit(boolean validateContent) throws Exception {
Map<String, String> writeOptions = new HashMap<>();
writeOptions.put(DataSourceWriteOptions.TABLE_NAME().key(), "test_table");
writeOptions.put("hoodie.table.name", "test_table");
writeOptions.put(DataSourceWriteOptions.TABLE_TYPE().key(), "COPY_ON_WRITE");
writeOptions.put(DataSourceWriteOptions.RECORDKEY_FIELD().key(), "_row_key");
writeOptions.put(DataSourceWriteOptions.PRECOMBINE_FIELD().key(), "timestamp");
writeOptions.put(DataSourceWriteOptions.OPERATION().key(), WriteOperationType.BULK_INSERT.value());
// Leave the metadata table off for the write; it is bootstrapped out of band below so that the
// data table's latest completed commit stays the instant the init instants are derived from.
writeOptions.put(HoodieMetadataConfig.ENABLE.key(), "false");
// Table version 6 is the one whose metadata table initialization instants carry the suffix.
writeOptions.put(HoodieWriteConfig.WRITE_TABLE_VERSION.key(), "6");

makeInsertDf("000", 50).write().format("hudi").options(writeOptions)
.mode(SaveMode.Overwrite)
.save(basePath);

HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()
.withPath(basePath)
.forTable("test_table")
.withWriteTableVersion(6)
.withMetadataConfig(HoodieMetadataConfig.newBuilder()
.enable(true)
.withEnableGlobalRecordLevelIndex(true)
.withRecordIndexFileGroupCount(1, 1)
.build())
.build();
HoodieTableMetaClient metaClientBeforeInit = HoodieTableMetaClient.builder()
.setBasePath(basePath).setConf(HadoopFSUtils.getStorageConf(jsc.hadoopConfiguration())).build();
assertEquals(HoodieTableVersion.SIX, metaClientBeforeInit.getTableConfig().getTableVersion(),
"the suffixed initialization instants only exist on table version 6");

// Creating the writer initializes the FILES and RECORD_INDEX partitions from the filesystem
// without adding a commit to the data table. Go through the factory so the table-version-6 writer
// is selected, exactly as production does.
try (HoodieTableMetadataWriter ignored = SparkMetadataWriterFactory.create(
HadoopFSUtils.getStorageConf(jsc.hadoopConfiguration()), writeConfig, context,
Option.empty(), metaClientBeforeInit.getTableConfig())) {
// constructing the writer performs the initialization
}

HoodieTableMetaClient dataMetaClient = HoodieTableMetaClient.reload(metaClientBeforeInit);
assertTrue(dataMetaClient.getTableConfig().isMetadataPartitionAvailable(MetadataPartitionType.RECORD_INDEX),
"record index should be registered on the data table");
String latestDataCommit = dataMetaClient.getActiveTimeline().getCommitsAndCompactionTimeline()
.filterCompletedInstants().lastInstant().get().requestedTime();
HoodieTableMetaClient mdtMetaClient = HoodieTableMetaClient.builder()
.setBasePath(HoodieTableMetadata.getMetadataTableBasePath(basePath))
.setConf(HadoopFSUtils.getStorageConf(jsc.hadoopConfiguration())).build();
List<String> mdtInstants = mdtMetaClient.getActiveTimeline().filterCompletedInstants()
.getInstantsAsStream().map(HoodieInstant::requestedTime).collect(Collectors.toList());
assertTrue(
mdtInstants.stream().allMatch(instant -> instant.startsWith(latestDataCommit)
&& instant.length() > latestDataCommit.length()),
"expected every metadata instant to be a suffixed extension of " + latestDataCommit
+ " but got " + mdtInstants);

HoodieMetadataTableValidator.Config config = new HoodieMetadataTableValidator.Config();
config.basePath = "file:" + basePath;
config.validateLatestFileSlices = true;
// validateRecordIndexContent shadows validateRecordIndexCount, so toggling it exercises both paths.
config.validateRecordIndexContent = validateContent;
config.validateRecordIndexCount = true;
config.ignoreFailed = true;

HoodieMetadataTableValidator validator = new HoodieMetadataTableValidator(jsc, config);
// Assert on the record index validation directly: doMetadataTableValidation() reports any
// non-HoodieValidationException as a successful run, so run() alone cannot distinguish a genuine
// pass from the read blowing up.
assertDoesNotThrow(() -> validator.validateRecordIndex(new HoodieSparkEngineContext(jsc), dataMetaClient),
"record index validation should pass against an intact record index");
assertTrue(validator.run(), "validation should succeed against an intact record index");
assertFalse(validator.hasValidationFailure(), () -> "unexpected validation failures: "
+ validator.getThrowables());
}

@Test
public void testMetadataTableInstantForIncludesEveryDerivedInstant() {
String dataTableInstant = "20231012054834279";

String queryInstant = HoodieMetadataTableValidator.metadataTableInstantFor(dataTableInstant);

assertEquals("20231012054834280", queryInstant, "the bound should advance the instant by one millisecond");
// 010-013 are appended when a metadata table partition is initialized, 001-006 by the
// metadata-table-internal operations; all of them must fall inside the queried window.
for (String suffix : new String[] {"001", "002", "003", "004", "005", "006", "010", "011", "012", "013"}) {
assertTrue(
InstantComparison.compareTimestamps(dataTableInstant + suffix, InstantComparison.LESSER_THAN_OR_EQUALS, queryInstant),
() -> "metadata instant " + dataTableInstant + suffix + " should be included by " + queryInstant);
}
// ... while the next data table instant stays outside it.
assertTrue(InstantComparison.compareTimestamps("20231012054834281", InstantComparison.GREATER_THAN, queryInstant),
"a later data table instant should not be included");
}

@Test
public void testMetadataTableInstantForLeavesNonTimestampInstantUnchanged() {
assertEquals("100", HoodieMetadataTableValidator.metadataTableInstantFor("100"));
}
}
Loading