You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@xtable.apache.org by ti...@apache.org on 2024/03/04 23:04:52 UTC
(incubator-xtable) branch main updated: Fix data file extraction from Hudi restore metadata with log files (#332)
This is an automated email from the ASF dual-hosted git repository.
timbrown 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 7577727 Fix data file extraction from Hudi restore metadata with log files (#332)
7577727 is described below
commit 7577727cf28117811bcc90b8c238254495b7dd26
Author: Y Ethan Guo <et...@gmail.com>
AuthorDate: Mon Mar 4 15:04:48 2024 -0800
Fix data file extraction from Hudi restore metadata with log files (#332)
---
.../io/onetable/hudi/HudiDataFileExtractor.java | 1 +
.../java/io/onetable/TestAbstractHudiTable.java | 50 ++++++++++++++++++++--
.../java/io/onetable/hudi/ITHudiSourceClient.java | 29 +++++++++++--
3 files changed, 73 insertions(+), 7 deletions(-)
diff --git a/core/src/main/java/io/onetable/hudi/HudiDataFileExtractor.java b/core/src/main/java/io/onetable/hudi/HudiDataFileExtractor.java
index 08fbc70..b22e743 100644
--- a/core/src/main/java/io/onetable/hudi/HudiDataFileExtractor.java
+++ b/core/src/main/java/io/onetable/hudi/HudiDataFileExtractor.java
@@ -271,6 +271,7 @@ public class HudiDataFileExtractor implements AutoCloseable {
throw new OneIOException("Unable to parse path " + path, e);
}
})
+ .filter(uri -> !FSUtils.isLogFile(new Path(uri).getName()))
.map(HoodieBaseFile::new)
.map(baseFile -> buildFileWithoutStats(partitionValues, baseFile))
.collect(toList(deletedPaths.size()));
diff --git a/core/src/test/java/io/onetable/TestAbstractHudiTable.java b/core/src/test/java/io/onetable/TestAbstractHudiTable.java
index 8fe5c1b..b3dfb10 100644
--- a/core/src/test/java/io/onetable/TestAbstractHudiTable.java
+++ b/core/src/test/java/io/onetable/TestAbstractHudiTable.java
@@ -22,6 +22,7 @@ import static io.onetable.hudi.HudiTestUtil.getHoodieWriteConfig;
import static org.apache.hudi.keygen.constant.KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME;
import static org.junit.jupiter.api.Assertions.assertAll;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.IOException;
import java.io.InputStream;
@@ -57,9 +58,9 @@ import org.apache.avro.generic.GenericRecord;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.LocalFileSystem;
-import org.junit.jupiter.api.Assertions;
import org.apache.hudi.avro.model.HoodieCleanMetadata;
+import org.apache.hudi.avro.model.HoodieRestoreMetadata;
import org.apache.hudi.client.BaseHoodieWriteClient;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.transaction.lock.InProcessLockProvider;
@@ -80,7 +81,9 @@ import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
import org.apache.hudi.common.model.WriteConcurrencyMode;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.marker.MarkerType;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
import org.apache.hudi.common.util.JsonUtils;
import org.apache.hudi.common.util.Option;
@@ -310,19 +313,27 @@ public abstract class TestAbstractHudiTable
public void clean() {
HoodieCleanMetadata metadata = getWriteClient().clean();
// Assert that files are deleted to ensure test is realistic
- Assertions.assertTrue(metadata.getTotalFilesDeleted() > 0);
+ assertTrue(metadata.getTotalFilesDeleted() > 0);
}
public void rollback(String commitInstant) {
getWriteClient().rollback(commitInstant);
}
- public void savepointRestoreForPreviousInstant() {
+ /**
+ * Restore to nth latest instant. 1 means restore to the previous instant, 2 would be the instant
+ * before that, and so on. 0 would mean restore to the latest instant which doesn't make sense to
+ * do.
+ *
+ * @param n instant to restore from
+ */
+ public void savepointRestoreFromNthMostRecentInstant(int n) {
List<HoodieInstant> commitInstants =
metaClient.getActiveTimeline().reload().getCommitsTimeline().getInstants();
- HoodieInstant instantToRestore = commitInstants.get(commitInstants.size() - 2);
+ HoodieInstant instantToRestore = commitInstants.get(commitInstants.size() - 1 - n);
getWriteClient().savepoint(instantToRestore.getTimestamp(), "user", "savepoint-test");
getWriteClient().restoreToSavepoint(instantToRestore.getTimestamp());
+ assertMergeOnReadRestoreContainsLogFiles();
}
public List<HoodieBaseFile> getAllLatestBaseFiles(HoodieTableFileSystemView fsView) {
@@ -334,6 +345,37 @@ public abstract class TestAbstractHudiTable
}
}
+ public void assertMergeOnReadRestoreContainsLogFiles() {
+ if (metaClient.getTableConfig().getTableType() == HoodieTableType.MERGE_ON_READ) {
+ HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline().reload();
+ HoodieInstant restoreInstant = activeTimeline.getRestoreTimeline().firstInstant().get();
+ Option<byte[]> instantDetails = activeTimeline.getInstantDetails(restoreInstant);
+ try {
+ HoodieRestoreMetadata instantMetadata =
+ TimelineMetadataUtils.deserializeAvroMetadata(
+ instantDetails.get(), HoodieRestoreMetadata.class);
+ assertTrue(
+ instantMetadata.getHoodieRestoreMetadata().values().stream()
+ .flatMap(
+ list ->
+ list.stream()
+ .flatMap(
+ rollbackMetadata ->
+ rollbackMetadata.getPartitionMetadata().values().stream()
+ .flatMap(
+ partitionMetadata ->
+ partitionMetadata
+ .getSuccessDeleteFiles()
+ .stream())))
+ .map(uri -> FSUtils.isLogFile(new org.apache.hadoop.fs.Path(uri).getName()))
+ .reduce((a, b) -> a || b)
+ .get());
+ } catch (IOException e) {
+ throw new UncheckedIOException("Cannot deserialize restore metadata", e);
+ }
+ }
+ }
+
public static void assertNoWriteErrors(List<WriteStatus> statuses) {
assertAll(
statuses.stream()
diff --git a/core/src/test/java/io/onetable/hudi/ITHudiSourceClient.java b/core/src/test/java/io/onetable/hudi/ITHudiSourceClient.java
index 0fa4a72..aabb76b 100644
--- a/core/src/test/java/io/onetable/hudi/ITHudiSourceClient.java
+++ b/core/src/test/java/io/onetable/hudi/ITHudiSourceClient.java
@@ -31,10 +31,13 @@ import java.time.Instant;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import java.util.stream.Stream;
import lombok.Builder;
@@ -416,12 +419,20 @@ public class ITHudiSourceClient {
String commitInstant1 = table.startCommit();
List<HoodieRecord<HoodieAvroPayload>> insertsForCommit1 = table.generateRecords(50);
table.insertRecordsWithCommitAlreadyStarted(insertsForCommit1, commitInstant1, true);
+ allBaseFilePaths.add(table.getAllLatestBaseFilePaths());
+ // This is the commit we're going to savepoint and restore to
table.insertRecords(50, true);
allBaseFilePaths.add(table.getAllLatestBaseFilePaths());
- // Restore to commitInstant2.
- table.savepointRestoreForPreviousInstant();
+ List<HoodieRecord<HoodieAvroPayload>> recordList = table.insertRecords(50, true);
+ Set<String> baseFilePaths = new HashSet<>(table.getAllLatestBaseFilePaths());
+ table.upsertRecords(recordList.subList(0, 20), true);
+ baseFilePaths.addAll(table.getAllLatestBaseFilePaths());
+ // Note that restore removes all the new base files added by these two commits
+ allBaseFilePaths.add(new ArrayList<>(baseFilePaths));
+
+ table.savepointRestoreFromNthMostRecentInstant(2);
allBaseFilePaths.add(table.getAllLatestBaseFilePaths());
table.insertRecords(50, true);
@@ -444,7 +455,19 @@ public class ITHudiSourceClient {
TableChange tableChange = hudiClient.getTableChangeForCommit(instant);
allTableChanges.add(tableChange);
}
- validateTableChanges(allBaseFilePaths, allTableChanges);
+
+ IntStream.range(0, allTableChanges.size() - 1)
+ .forEach(
+ i -> {
+ if (i == 1) {
+ // Savepoint: no change
+ validateTableChange(
+ allBaseFilePaths.get(i), allBaseFilePaths.get(i), allTableChanges.get(i));
+ } else {
+ validateTableChange(
+ allBaseFilePaths.get(i), allBaseFilePaths.get(i + 1), allTableChanges.get(i));
+ }
+ });
}
}