Skip to content

[HUDI-6389] Fix instant time check against the active timeline in meta sync #8991

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 2 commits into from
Jun 16, 2023
Merged

[HUDI-6389] Fix instant time check against the active timeline in meta sync #8991

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
da5187e
[HUDI-6389] Fix instant time check against the active timeline in met…
yihua Jun 15, 2023
707d4cd
Fix test
yihua Jun 16, 2023
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
8 changes: 4 additions & 4 deletions hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,11 +232,11 @@ public static HoodieDefaultTimeline getTimeline(HoodieTableMetaClient metaClient
*/
public static HoodieTimeline getCommitsTimelineAfter(
HoodieTableMetaClient metaClient, String exclusiveStartInstantTime, Option<String> lastMaxCompletionTime) {
HoodieDefaultTimeline activeTimeline = metaClient.getActiveTimeline();
HoodieDefaultTimeline writeTimeline = metaClient.getActiveTimeline().getWriteTimeline();

HoodieDefaultTimeline timeline = activeTimeline.isBeforeTimelineStarts(exclusiveStartInstantTime)
? metaClient.getArchivedTimeline(exclusiveStartInstantTime).mergeTimeline(activeTimeline)
: activeTimeline;
HoodieDefaultTimeline timeline = writeTimeline.isBeforeTimelineStarts(exclusiveStartInstantTime)
? metaClient.getArchivedTimeline(exclusiveStartInstantTime).mergeTimeline(writeTimeline)
: writeTimeline;

HoodieDefaultTimeline timelineSinceLastSync = (HoodieDefaultTimeline) timeline.getCommitsTimeline()
.findInstantsAfter(exclusiveStartInstantTime, Integer.MAX_VALUE);
Expand Down
27 changes: 27 additions & 0 deletions hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -323,6 +323,33 @@ public void testGetCommitsTimelineAfter() throws IOException {
new HoodieInstant(COMPLETED, COMMIT_ACTION, "012", "012")),
TimelineUtils.getCommitsTimelineAfter(mockMetaClient, startTs, Option.of(startTs)));
verify(mockMetaClient, times(1)).getArchivedTimeline(any());

// Should load both archived and active timeline
startTs = "005";
mockMetaClient = prepareMetaClient(
Arrays.asList(
new HoodieInstant(COMPLETED, ROLLBACK_ACTION, "003", "003"),
new HoodieInstant(COMPLETED, ROLLBACK_ACTION, "007", "007"),
new HoodieInstant(COMPLETED, ROLLBACK_ACTION, "009", "009"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "010", "010"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "011", "011"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "012", "012")),
Arrays.asList(new HoodieInstant(COMPLETED, COMMIT_ACTION, "001", "001"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "002", "002"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "005", "005"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "006", "006"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "008", "008")),
startTs
);
verifyTimeline(
Arrays.asList(
new HoodieInstant(COMPLETED, COMMIT_ACTION, "006", "006"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "008", "008"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "010", "010"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "011", "011"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "012", "012")),
TimelineUtils.getCommitsTimelineAfter(mockMetaClient, startTs, Option.of(startTs)));
verify(mockMetaClient, times(1)).getArchivedTimeline(any());
}

private HoodieTableMetaClient prepareMetaClient(
Expand Down
36 changes: 36 additions & 0 deletions hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHiveSyncTool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.hudi.common.model.HoodieSyncTableStrategy;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator;
import org.apache.hudi.common.testutils.NetworkTestUtils;
import org.apache.hudi.common.testutils.SchemaTestUtil;
Expand Down Expand Up @@ -390,6 +391,41 @@ public void testBasicSync(boolean useSchemaFromCommitMetadata, String syncMode,
tablePartitions = hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME);
assertEquals(Option.of("300"), hiveClient.getLastCommitTimeSynced(HiveTestUtil.TABLE_NAME));
assertEquals(7, tablePartitions.size());

// Add the following instants to the active timeline and sync: "400" (rollback), "500" (commit)
// Last commit time sync is "500" after Hive sync
HiveTestUtil.addRollbackInstantToTable("400", "350");
HiveTestUtil.commitToTable("500", 7, useSchemaFromCommitMetadata);
reInitHiveSyncClient();
reSyncHiveTable();
tablePartitions = hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME);
assertEquals(Option.of("500"), hiveClient.getLastCommitTimeSynced(HiveTestUtil.TABLE_NAME));
assertEquals(8, tablePartitions.size());

// Add more instants with adding a partition and simulate the case where the commit adding
// the new partition is archived.
// Before simulated archival: "300", "400" (rollback), "500", "600" (adding new partition), "700", "800"
// After simulated archival: "400" (rollback), "700", "800"
// In this case, listing all partitions should be triggered to catch up.
HiveTestUtil.commitToTable("600", 8, useSchemaFromCommitMetadata);
HiveTestUtil.commitToTable("700", 1, useSchemaFromCommitMetadata);
HiveTestUtil.commitToTable("800", 1, useSchemaFromCommitMetadata);
HiveTestUtil.removeCommitFromActiveTimeline("300", COMMIT_ACTION);
HiveTestUtil.removeCommitFromActiveTimeline("500", COMMIT_ACTION);
HiveTestUtil.removeCommitFromActiveTimeline("600", COMMIT_ACTION);
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder()
.setConf(hiveClient.config.getHadoopConf()).setBasePath(basePath).build();
assertEquals(
Arrays.asList("400", "700", "800"),
metaClient.getActiveTimeline().getInstants().stream()
.map(HoodieInstant::getTimestamp).sorted()
.collect(Collectors.toList()));

reInitHiveSyncClient();
reSyncHiveTable();
tablePartitions = hiveClient.getAllPartitions(HiveTestUtil.TABLE_NAME);
assertEquals(Option.of("800"), hiveClient.getLastCommitTimeSynced(HiveTestUtil.TABLE_NAME));
assertEquals(9, tablePartitions.size());
}

@ParameterizedTest
Expand Down
71 changes: 50 additions & 21 deletions hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/testutils/HiveTestUtil.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.hive.testutils;

import org.apache.hudi.avro.HoodieAvroWriteSupport;
import org.apache.hudi.avro.model.HoodieRollbackMetadata;
import org.apache.hudi.common.bloom.BloomFilter;
import org.apache.hudi.common.bloom.BloomFilterFactory;
import org.apache.hudi.common.bloom.BloomFilterTypeCode;
Expand Down Expand Up @@ -81,6 +82,7 @@
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
Expand All @@ -92,6 +94,7 @@
import java.util.stream.Collectors;

import static org.apache.hudi.common.table.HoodieTableMetaClient.METAFOLDER_NAME;
import static org.apache.hudi.common.table.timeline.TimelineMetadataUtils.serializeRollbackMetadata;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_BATCH_SYNC_PARTITION_NUM;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_PASS;
import static org.apache.hudi.hive.HiveSyncConfigHolder.HIVE_URL;
Expand Down Expand Up @@ -263,6 +266,32 @@ public static void createReplaceCommit(String instantTime, String partitions, Wr
createReplaceCommitFile(replaceCommitMetadata, instantTime);
}

public static void addRollbackInstantToTable(String instantTime, String commitToRollback)
throws IOException {
HoodieRollbackMetadata rollbackMetadata = HoodieRollbackMetadata.newBuilder()
.setVersion(1)
.setStartRollbackTime(instantTime)
.setTotalFilesDeleted(1)
.setTimeTakenInMillis(1000)
.setCommitsRollback(Collections.singletonList(commitToRollback))
.setPartitionMetadata(Collections.emptyMap())
.setInstantsRollback(Collections.emptyList())
.build();

createMetaFile(
basePath,
HoodieTimeline.makeRequestedRollbackFileName(instantTime),
"".getBytes());
createMetaFile(
basePath,
HoodieTimeline.makeInflightRollbackFileName(instantTime),
"".getBytes());
createMetaFile(
basePath,
HoodieTimeline.makeRollbackFileName(instantTime),
serializeRollbackMetadata(rollbackMetadata).get());
}

public static void createCOWTableWithSchema(String instantTime, String schemaFileName)
throws IOException, URISyntaxException {
Path path = new Path(basePath);
Expand Down Expand Up @@ -521,21 +550,17 @@ private static void checkResult(boolean result) {
}

public static void createCommitFile(HoodieCommitMetadata commitMetadata, String instantTime, String basePath) throws IOException {
byte[] bytes = commitMetadata.toJsonString().getBytes(StandardCharsets.UTF_8);
Path fullPath = new Path(basePath + "/" + METAFOLDER_NAME + "/"
+ HoodieTimeline.makeCommitFileName(instantTime));
FSDataOutputStream fsout = fileSystem.create(fullPath, true);
fsout.write(bytes);
fsout.close();
createMetaFile(
basePath,
HoodieTimeline.makeCommitFileName(instantTime),
commitMetadata.toJsonString().getBytes(StandardCharsets.UTF_8));
}

public static void createReplaceCommitFile(HoodieReplaceCommitMetadata commitMetadata, String instantTime) throws IOException {
byte[] bytes = commitMetadata.toJsonString().getBytes(StandardCharsets.UTF_8);
Path fullPath = new Path(basePath + "/" + METAFOLDER_NAME + "/"
+ HoodieTimeline.makeReplaceFileName(instantTime));
FSDataOutputStream fsout = fileSystem.create(fullPath, true);
fsout.write(bytes);
fsout.close();
createMetaFile(
basePath,
HoodieTimeline.makeReplaceFileName(instantTime),
commitMetadata.toJsonString().getBytes(StandardCharsets.UTF_8));
}

public static void createCommitFileWithSchema(HoodieCommitMetadata commitMetadata, String instantTime, boolean isSimpleSchema) throws IOException {
Expand All @@ -545,19 +570,23 @@ public static void createCommitFileWithSchema(HoodieCommitMetadata commitMetadat

private static void createCompactionCommitFile(HoodieCommitMetadata commitMetadata, String instantTime)
throws IOException {
byte[] bytes = commitMetadata.toJsonString().getBytes(StandardCharsets.UTF_8);
Path fullPath = new Path(basePath + "/" + METAFOLDER_NAME + "/"
+ HoodieTimeline.makeCommitFileName(instantTime));
FSDataOutputStream fsout = fileSystem.create(fullPath, true);
fsout.write(bytes);
fsout.close();
createMetaFile(
basePath,
HoodieTimeline.makeCommitFileName(instantTime),
commitMetadata.toJsonString().getBytes(StandardCharsets.UTF_8));
}

private static void createDeltaCommitFile(HoodieCommitMetadata deltaCommitMetadata, String deltaCommitTime)
throws IOException {
byte[] bytes = deltaCommitMetadata.toJsonString().getBytes(StandardCharsets.UTF_8);
Path fullPath = new Path(basePath + "/" + METAFOLDER_NAME + "/"
+ HoodieTimeline.makeDeltaFileName(deltaCommitTime));
createMetaFile(
basePath,
HoodieTimeline.makeDeltaFileName(deltaCommitTime),
deltaCommitMetadata.toJsonString().getBytes(StandardCharsets.UTF_8));
}

private static void createMetaFile(String basePath, String fileName, byte[] bytes)
throws IOException {
Path fullPath = new Path(basePath + "/" + METAFOLDER_NAME + "/" + fileName);
FSDataOutputStream fsout = fileSystem.create(fullPath, true);
fsout.write(bytes);
fsout.close();
Expand Down