From 7d2d385261f3c1cb4dfe8c6cca617ea01e10a8e6 Mon Sep 17 00:00:00 2001 From: Vinoth Chandar Date: Thu, 7 Jan 2021 10:58:31 -0800 Subject: [PATCH 1/9] Scratch pR for HUDI-1502 --- .../org/apache/hudi/io/HoodieAppendHandle.java | 1 + .../BaseMergeOnReadRollbackActionExecutor.java | 2 +- .../src/main/avro/HoodieRollbackMetadata.avsc | 16 +++++++++++----- .../apache/hudi/common/HoodieRollbackStat.java | 1 + .../table/timeline/TimelineMetadataUtils.java | 4 ++-- .../hudi/metadata/HoodieTableMetadataUtil.java | 6 +++--- 6 files changed, 19 insertions(+), 11 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java index 3faac2e6ed7f7..8f272b2c5ffc4 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java @@ -164,6 +164,7 @@ private void init(HoodieRecord record) { // Since the actual log file written to can be different based on when rollover happens, we use the // base file to denote some log appends happened on a slice. writeToken will still fence concurrent // writers. + // TODO? are these sufficient? createMarkerFile(partitionPath, FSUtils.makeDataFileName(baseInstantTime, writeToken, fileId, hoodieTable.getBaseFileExtension())); this.writer = createLogWriter(fileSlice, baseInstantTime); diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseMergeOnReadRollbackActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseMergeOnReadRollbackActionExecutor.java index 44cd5c8cad6ae..22cafba800ee3 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseMergeOnReadRollbackActionExecutor.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseMergeOnReadRollbackActionExecutor.java @@ -57,7 +57,7 @@ public BaseMergeOnReadRollbackActionExecutor(HoodieEngineContext context, } @Override - protected List executeRollback() throws IOException { + protected List executeRollback() { HoodieTimer rollbackTimer = new HoodieTimer(); rollbackTimer.startTimer(); diff --git a/hudi-common/src/main/avro/HoodieRollbackMetadata.avsc b/hudi-common/src/main/avro/HoodieRollbackMetadata.avsc index 069881ea113c2..e354d5ccf1efc 100644 --- a/hudi-common/src/main/avro/HoodieRollbackMetadata.avsc +++ b/hudi-common/src/main/avro/HoodieRollbackMetadata.avsc @@ -31,18 +31,24 @@ {"name": "partitionPath", "type": "string"}, {"name": "successDeleteFiles", "type": {"type": "array", "items": "string"}}, {"name": "failedDeleteFiles", "type": {"type": "array", "items": "string"}}, - {"name": "appendFiles", "type": { + {"name": "rollbackLogFiles", "type": { "type": "map", - "doc": "Files to which append blocks were written", + "doc": "Files to which rollback command blocks were written", + "values": { + "type": "long", + "doc": "Size of this file in bytes" + } + }}, + {"name": "writtenLogFiles", "type": { + "type": "map", + "doc": "All log files written to by the rolled back commits", "values": { "type": "long", "doc": "Size of this file in bytes" } }} ] - } - } - }, + }}}, { "name":"version", "type":["int", "null"], diff --git a/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java b/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java index a3191fa026c84..e994cc6e42253 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java @@ -38,6 +38,7 @@ public class HoodieRollbackStat implements Serializable { private final List failedDeleteFiles; // Count of HoodieLogFile to commandBlocks written for a particular rollback private final Map commandBlocksCount; + private final Map logFilesWrittenToSize; public HoodieRollbackStat(String partitionPath, List successDeleteFiles, List failedDeleteFiles, Map commandBlocksCount) { diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java index 32e60c30bd868..7eccc0c648213 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java @@ -70,10 +70,10 @@ public static HoodieRollbackMetadata convertRollbackMetadata(String startRollbac Map partitionMetadataBuilder = new HashMap<>(); int totalDeleted = 0; for (HoodieRollbackStat stat : rollbackStats) { - Map appendFiles = stat.getCommandBlocksCount().keySet().stream() + Map rollbackLogFiles = stat.getCommandBlocksCount().keySet().stream() .collect(Collectors.toMap(f -> f.getPath().toString(), FileStatus::getLen)); HoodieRollbackPartitionMetadata metadata = new HoodieRollbackPartitionMetadata(stat.getPartitionPath(), - stat.getSuccessDeleteFiles(), stat.getFailedDeleteFiles(), appendFiles); + stat.getSuccessDeleteFiles(), stat.getFailedDeleteFiles(), rollbackLogFiles, Collections.EMPTY_MAP); partitionMetadataBuilder.put(stat.getPartitionPath(), metadata); totalDeleted += stat.getSuccessDeleteFiles().size(); } 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 115001a02a3a7..79816fafc1405 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 @@ -239,7 +239,7 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad rollbackMetadata.getPartitionMetadata().values().forEach(pm -> { // Has this rollback produced new files? - boolean hasAppendFiles = pm.getAppendFiles().values().stream().mapToLong(Long::longValue).sum() > 0; + boolean hasAppendFiles = pm.getRollbackLogFiles().values().stream().mapToLong(Long::longValue).sum() > 0; // If commit being rolled back has not been synced to metadata table yet then there is no need to update metadata boolean shouldSkip = lastSyncTs.isPresent() && HoodieTimeline.compareTimestamps(rollbackMetadata.getCommitsRollback().get(0), HoodieTimeline.GREATER_THAN, lastSyncTs.get()); @@ -262,13 +262,13 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad partitionToDeletedFiles.get(partition).addAll(deletedFiles); } - if (!pm.getAppendFiles().isEmpty()) { + if (!pm.getRollbackLogFiles().isEmpty()) { if (!partitionToAppendedFiles.containsKey(partition)) { partitionToAppendedFiles.put(partition, new HashMap<>()); } // Extract appended file name from the absolute paths saved in getAppendFiles() - pm.getAppendFiles().forEach((path, size) -> { + pm.getRollbackLogFiles().forEach((path, size) -> { partitionToAppendedFiles.get(partition).merge(new Path(path).getName(), size, (oldSize, newSizeCopy) -> { return size + oldSize; }); From 2bec2d345f61059d66b47681fdc93cf1879b08c5 Mon Sep 17 00:00:00 2001 From: Sivabalan Narayanan Date: Fri, 8 Jan 2021 03:57:54 -0500 Subject: [PATCH 2/9] [HUDI-1502] Fixing rollback interplay with metadata sync up --- .../AbstractMarkerBasedRollbackStrategy.java | 26 ++++++++++++--- ...BaseMergeOnReadRollbackActionExecutor.java | 1 - .../table/action/rollback/RollbackUtils.java | 7 ++-- .../rollback/ListingBasedRollbackHelper.java | 29 ++++++++++++++--- .../SparkMarkerBasedRollbackStrategy.java | 30 ++++++++++++++++- .../metadata/TestHoodieBackedMetadata.java | 32 +++++++++++-------- .../hudi/common/HoodieRollbackStat.java | 28 ++++++++++++++-- .../table/timeline/TimelineMetadataUtils.java | 4 ++- .../metadata/HoodieTableMetadataUtil.java | 13 ++++++++ .../hudi/common/table/TestTimelineUtils.java | 3 +- 10 files changed, 143 insertions(+), 30 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java index cb6fff936716e..fb6df5a227bbe 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java @@ -32,6 +32,7 @@ import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.table.HoodieTable; + import org.apache.log4j.LogManager; import org.apache.log4j.Logger; @@ -53,9 +54,9 @@ public abstract class AbstractMarkerBasedRollbackStrategy table, HoodieEngineContext context, HoodieWriteConfig config, String instantTime) { this.table = table; @@ -90,6 +91,7 @@ protected HoodieRollbackStat undoAppend(String appendBaseFilePath, HoodieInstant String fileId = FSUtils.getFileIdFromFilePath(baseFilePathForAppend); String baseCommitTime = FSUtils.getCommitTime(baseFilePathForAppend.getName()); String partitionPath = FSUtils.getRelativePartitionPath(new Path(basePath), new Path(basePath, appendBaseFilePath).getParent()); + final Map probableLogFileMap = getProbableFileSizeMap(partitionPath, baseCommitTime); HoodieLogFormat.Writer writer = null; try { @@ -129,9 +131,23 @@ protected HoodieRollbackStat undoAppend(String appendBaseFilePath, HoodieInstant 1L); } - return HoodieRollbackStat.newBuilder() + HoodieRollbackStat.Builder builder = HoodieRollbackStat.newBuilder() .withPartitionPath(partitionPath) - .withRollbackBlockAppendResults(filesToNumBlocksRollback) - .build(); + .withRollbackBlockAppendResults(filesToNumBlocksRollback); + if (probableLogFileMap != null) { + builder.withProbableLogFileToSizeMap(probableLogFileMap); + } + return builder.build(); + } + + /** + * Returns probable log files for the respective baseCommitTime to assist in metadata table syncing. + * @param partitionPath partition path of interest + * @param baseCommitTime base commit time of interest + * @return Map + * @throws IOException + */ + protected Map getProbableFileSizeMap(String partitionPath, String baseCommitTime) throws IOException { + return null; } } diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseMergeOnReadRollbackActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseMergeOnReadRollbackActionExecutor.java index 22cafba800ee3..2e751443abc00 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseMergeOnReadRollbackActionExecutor.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseMergeOnReadRollbackActionExecutor.java @@ -28,7 +28,6 @@ import org.apache.log4j.LogManager; import org.apache.log4j.Logger; -import java.io.IOException; import java.util.ArrayList; import java.util.List; diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java index 18f284ea35efd..96c99c7c89f92 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java @@ -74,14 +74,16 @@ static HoodieRollbackStat mergeRollbackStat(HoodieRollbackStat stat1, HoodieRoll final List successDeleteFiles = new ArrayList<>(); final List failedDeleteFiles = new ArrayList<>(); final Map commandBlocksCount = new HashMap<>(); - final List filesToRollback = new ArrayList<>(); + final Map probableLogFileMap = new HashMap<>(); Option.ofNullable(stat1.getSuccessDeleteFiles()).ifPresent(successDeleteFiles::addAll); Option.ofNullable(stat2.getSuccessDeleteFiles()).ifPresent(successDeleteFiles::addAll); Option.ofNullable(stat1.getFailedDeleteFiles()).ifPresent(failedDeleteFiles::addAll); Option.ofNullable(stat2.getFailedDeleteFiles()).ifPresent(failedDeleteFiles::addAll); Option.ofNullable(stat1.getCommandBlocksCount()).ifPresent(commandBlocksCount::putAll); Option.ofNullable(stat2.getCommandBlocksCount()).ifPresent(commandBlocksCount::putAll); - return new HoodieRollbackStat(stat1.getPartitionPath(), successDeleteFiles, failedDeleteFiles, commandBlocksCount); + Option.ofNullable(stat1.getProbableLogFileToSizeMap()).ifPresent(probableLogFileMap::putAll); + Option.ofNullable(stat2.getProbableLogFileToSizeMap()).ifPresent(probableLogFileMap::putAll); + return new HoodieRollbackStat(stat1.getPartitionPath(), successDeleteFiles, failedDeleteFiles, commandBlocksCount, probableLogFileMap); } /** @@ -121,6 +123,7 @@ public static List generateRollbackRequestsUsingFil HoodieActiveTimeline activeTimeline = table.getMetaClient().reloadActiveTimeline(); List partitionRollbackRequests = new ArrayList<>(); switch (instantToRollback.getAction()) { + case HoodieTimeline.COMMIT_ACTION: LOG.info("Rolling back commit action."); partitionRollbackRequests.add( diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java index a614aebfd78ae..5c048c4d436bb 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java @@ -73,6 +73,7 @@ public ListingBasedRollbackHelper(HoodieTableMetaClient metaClient, HoodieWriteC public List performRollback(HoodieEngineContext context, HoodieInstant instantToRollback, List rollbackRequests) { int sparkPartitions = Math.max(Math.min(rollbackRequests.size(), config.getRollbackParallelism()), 1); context.setJobStatus(this.getClass().getSimpleName(), "Perform rollback actions"); + JavaPairRDD partitionPathRollbackStatsPairRDD = maybeDeleteAndCollectStats(context, instantToRollback, rollbackRequests, sparkPartitions, true); return partitionPathRollbackStatsPairRDD.reduceByKey(RollbackUtils::mergeRollbackStat).map(Tuple2::_2).collect(); } @@ -98,8 +99,9 @@ public List collectRollbackStats(HoodieEngineContext context * @return stats collected with or w/o actual deletions. */ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineContext context, HoodieInstant instantToRollback, List rollbackRequests, - int sparkPartitions, boolean doDelete) { + int sparkPartitions, boolean doDelete) { JavaSparkContext jsc = HoodieSparkEngineContext.getSparkContext(context); + return jsc.parallelize(rollbackRequests, sparkPartitions).mapToPair(rollbackRequest -> { switch (rollbackRequest.getType()) { case DELETE_DATA_FILES_ONLY: { @@ -116,6 +118,24 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC .withDeletedFileResults(filesToDeletedStatus).build()); } case APPEND_ROLLBACK_BLOCK: { + // collect all log files that is supposed to be deleted with this rollback + String baseCommit = rollbackRequest.getLatestBaseInstant().get(); + SerializablePathFilter filter = (path) -> { + if (FSUtils.isLogFile(path)) { + // Since the baseCommitTime is the only commit for new log files, it's okay here + String fileCommitTime = FSUtils.getBaseCommitTimeFromLogPath(path); + return baseCommit.equals(fileCommitTime); + } + return false; + }; + + final Map probableLogFileMap = new HashMap<>(); + FileSystem fs = metaClient.getFs(); + FileStatus[] probableLogFiles = fs.listStatus(FSUtils.getPartitionPath(config.getBasePath(), rollbackRequest.getPartitionPath()), filter); + for (FileStatus fileStatus : probableLogFiles) { + probableLogFileMap.put(fileStatus, fileStatus.getLen()); + } + Writer writer = null; try { writer = HoodieLogFormat.newWriterBuilder() @@ -123,13 +143,13 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC .withFileId(rollbackRequest.getFileId().get()) .overBaseCommit(rollbackRequest.getLatestBaseInstant().get()).withFs(metaClient.getFs()) .withFileExtension(HoodieLogFile.DELTA_EXTENSION).build(); - // generate metadata if (doDelete) { Map header = generateHeader(instantToRollback.getTimestamp()); // if update belongs to an existing log file writer.appendBlock(new HoodieCommandBlock(header)); } + } catch (IOException | InterruptedException io) { throw new HoodieRollbackException("Failed to rollback for instant " + instantToRollback, io); } finally { @@ -149,9 +169,11 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC metaClient.getFs().getFileStatus(Objects.requireNonNull(writer).getLogFile().getPath()), 1L ); + return new Tuple2<>(rollbackRequest.getPartitionPath(), HoodieRollbackStat.newBuilder().withPartitionPath(rollbackRequest.getPartitionPath()) - .withRollbackBlockAppendResults(filesToNumBlocksRollback).build()); + .withRollbackBlockAppendResults(filesToNumBlocksRollback) + .withProbableLogFileToSizeMap(probableLogFileMap).build()); } default: throw new IllegalStateException("Unknown Rollback action " + rollbackRequest); @@ -159,7 +181,6 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC }); } - /** * Common method used for cleaning out base files under a partition path during rollback of a set of commits. */ diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java index 0598e54c04c76..d7608fd2c8e27 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java @@ -22,6 +22,7 @@ import org.apache.hudi.client.common.HoodieSparkEngineContext; import org.apache.hudi.common.HoodieRollbackStat; import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.fs.FSUtils; import org.apache.hudi.common.model.HoodieKey; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieRecordPayload; @@ -31,17 +32,25 @@ import org.apache.hudi.exception.HoodieRollbackException; import org.apache.hudi.table.HoodieTable; import org.apache.hudi.table.MarkerFiles; +import org.apache.hudi.table.action.rollback.ListingBasedRollbackHelper.SerializablePathFilter; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; +import java.io.IOException; +import java.util.HashMap; import java.util.List; +import java.util.Map; import scala.Tuple2; @SuppressWarnings("checkstyle:LineLength") public class SparkMarkerBasedRollbackStrategy extends AbstractMarkerBasedRollbackStrategy>, JavaRDD, JavaRDD> { - public SparkMarkerBasedRollbackStrategy(HoodieTable>, JavaRDD, JavaRDD> table, HoodieEngineContext context, HoodieWriteConfig config, String instantTime) { + + public SparkMarkerBasedRollbackStrategy(HoodieTable>, JavaRDD, JavaRDD> table, HoodieEngineContext context, HoodieWriteConfig config, + String instantTime) { super(table, context, config, instantTime); } @@ -75,4 +84,23 @@ public List execute(HoodieInstant instantToRollback) { throw new HoodieRollbackException("Error rolling back using marker files written for " + instantToRollback, e); } } + + protected Map getProbableFileSizeMap(String partitionPath, String baseCommitTime) throws IOException { + // collect all log files that is supposed to be deleted with this rollback + SerializablePathFilter filter = (path) -> { + if (FSUtils.isLogFile(path)) { + String fileCommitTime = FSUtils.getBaseCommitTimeFromLogPath(path); + return baseCommitTime.equals(fileCommitTime); + } + return false; + }; + + final Map probableLogFileMap = new HashMap<>(); + FileSystem fs = table.getMetaClient().getFs(); + FileStatus[] toBeDeleted = fs.listStatus(FSUtils.getPartitionPath(config.getBasePath(), partitionPath), filter); + for (FileStatus fileStatus : toBeDeleted) { + probableLogFileMap.put(fileStatus, fileStatus.getLen()); + } + return probableLogFileMap; + } } diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java index 32cec71feafc9..ed9e85b9dc875 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java @@ -78,6 +78,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; public class TestHoodieBackedMetadata extends HoodieClientTestHarness { + private static final Logger LOG = LogManager.getLogger(TestHoodieBackedMetadata.class); @TempDir @@ -259,13 +260,11 @@ public void testTableOperations(HoodieTableType tableType) throws Exception { /** * Test rollback of various table operations sync to Metadata Table correctly. */ - //@ParameterizedTest - //@EnumSource(HoodieTableType.class) - //public void testRollbackOperations(HoodieTableType tableType) throws Exception { - @Test - public void testRollbackOperations() throws Exception { + @ParameterizedTest + @EnumSource(HoodieTableType.class) + public void testRollbackOperations(HoodieTableType tableType) throws Exception { //FIXME(metadata): This is broken for MOR, until HUDI-1434 is fixed - init(HoodieTableType.COPY_ON_WRITE); + init(tableType); HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc); try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext, getWriteConfig(true, true))) { @@ -369,9 +368,7 @@ public void testRollbackOperations() throws Exception { } /** - * Test when syncing rollback to metadata if the commit being rolled back has not been synced that essentially a no-op - * occurs to metadata. - * @throws Exception + * Test when syncing rollback to metadata if the commit being rolled back has not been synced that essentially a no-op occurs to metadata. */ @Test public void testRollbackUnsyncedCommit() throws Exception { @@ -517,8 +514,7 @@ public void testSync(HoodieTableType tableType) throws Exception { } /** - * Instants on Metadata Table should be archived as per config. - * Metadata Table should be automatically compacted as per config. + * Instants on Metadata Table should be archived as per config. Metadata Table should be automatically compacted as per config. */ @Test public void testCleaningArchivingAndCompaction() throws Exception { @@ -741,8 +737,6 @@ public void testMetadataOutOfSync() throws Exception { /** * Validate the metadata tables contents to ensure it matches what is on the file system. - * - * @throws IOException */ private void validateMetadata(SparkRDDWriteClient client) throws IOException { HoodieWriteConfig config = client.getConfig(); @@ -794,7 +788,19 @@ private void validateMetadata(SparkRDDWriteClient client) throws IOException { if ((fsFileNames.size() != metadataFilenames.size()) || (!fsFileNames.equals(metadataFilenames))) { LOG.info("*** File system listing = " + Arrays.toString(fsFileNames.toArray())); LOG.info("*** Metadata listing = " + Arrays.toString(metadataFilenames.toArray())); + + for (String fileName : fsFileNames) { + if (!metadataFilenames.contains(fileName)) { + System.out.println(partition + "FsFilename " + fileName + " not found in Meta data"); + } + } + for (String fileName : metadataFilenames) { + if (!fsFileNames.contains(fileName)) { + System.out.println(partition + "Metadata file " + fileName + " not found in original FS"); + } + } } + assertEquals(fsFileNames.size(), metadataFilenames.size(), "Files within partition " + partition + " should match"); assertTrue(fsFileNames.equals(metadataFilenames), "Files within partition " + partition + " should match"); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java b/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java index e994cc6e42253..4951ba63946e9 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java @@ -38,7 +38,8 @@ public class HoodieRollbackStat implements Serializable { private final List failedDeleteFiles; // Count of HoodieLogFile to commandBlocks written for a particular rollback private final Map commandBlocksCount; - private final Map logFilesWrittenToSize; + // all log files with same base instant as instant to be rolledback + private final Map probableLogFileToSizeMap; public HoodieRollbackStat(String partitionPath, List successDeleteFiles, List failedDeleteFiles, Map commandBlocksCount) { @@ -46,6 +47,16 @@ public HoodieRollbackStat(String partitionPath, List successDeleteFiles, this.successDeleteFiles = successDeleteFiles; this.failedDeleteFiles = failedDeleteFiles; this.commandBlocksCount = commandBlocksCount; + this.probableLogFileToSizeMap = Collections.EMPTY_MAP; + } + + public HoodieRollbackStat(String partitionPath, List successDeleteFiles, List failedDeleteFiles, + Map commandBlocksCount, Map probableLogFileToSizeMap) { + this.partitionPath = partitionPath; + this.successDeleteFiles = successDeleteFiles; + this.failedDeleteFiles = failedDeleteFiles; + this.commandBlocksCount = commandBlocksCount; + this.probableLogFileToSizeMap = probableLogFileToSizeMap; } public Map getCommandBlocksCount() { @@ -64,6 +75,10 @@ public List getFailedDeleteFiles() { return failedDeleteFiles; } + public Map getProbableLogFileToSizeMap() { + return probableLogFileToSizeMap; + } + public static HoodieRollbackStat.Builder newBuilder() { return new Builder(); } @@ -76,6 +91,7 @@ public static class Builder { private List successDeleteFiles; private List failedDeleteFiles; private Map commandBlocksCount; + private Map probableLogFileToSizeMap; private String partitionPath; public Builder withDeletedFileResults(Map deletedFiles) { @@ -101,6 +117,11 @@ public Builder withRollbackBlockAppendResults(Map commandBlock return this; } + public Builder withProbableLogFileToSizeMap(Map probableLogFileToSizeMap) { + this.probableLogFileToSizeMap = probableLogFileToSizeMap; + return this; + } + public Builder withPartitionPath(String partitionPath) { this.partitionPath = partitionPath; return this; @@ -116,7 +137,10 @@ public HoodieRollbackStat build() { if (commandBlocksCount == null) { commandBlocksCount = Collections.EMPTY_MAP; } - return new HoodieRollbackStat(partitionPath, successDeleteFiles, failedDeleteFiles, commandBlocksCount); + if (probableLogFileToSizeMap == null) { + probableLogFileToSizeMap = Collections.EMPTY_MAP; + } + return new HoodieRollbackStat(partitionPath, successDeleteFiles, failedDeleteFiles, commandBlocksCount, probableLogFileToSizeMap); } } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java index 7eccc0c648213..c29cf6829db5e 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java @@ -72,8 +72,10 @@ public static HoodieRollbackMetadata convertRollbackMetadata(String startRollbac for (HoodieRollbackStat stat : rollbackStats) { Map rollbackLogFiles = stat.getCommandBlocksCount().keySet().stream() .collect(Collectors.toMap(f -> f.getPath().toString(), FileStatus::getLen)); + Map probableLogFiles = stat.getProbableLogFileToSizeMap().keySet().stream() + .collect(Collectors.toMap(f -> f.getPath().toString(), FileStatus::getLen)); HoodieRollbackPartitionMetadata metadata = new HoodieRollbackPartitionMetadata(stat.getPartitionPath(), - stat.getSuccessDeleteFiles(), stat.getFailedDeleteFiles(), rollbackLogFiles, Collections.EMPTY_MAP); + stat.getSuccessDeleteFiles(), stat.getFailedDeleteFiles(), rollbackLogFiles, probableLogFiles); partitionMetadataBuilder.put(stat.getPartitionPath(), metadata); totalDeleted += stat.getSuccessDeleteFiles().size(); } 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 79816fafc1405..a797485521f6d 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 @@ -274,6 +274,19 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad }); }); } + + if (!pm.getWrittenLogFiles().isEmpty()) { + if (!partitionToAppendedFiles.containsKey(partition)) { + partitionToAppendedFiles.put(partition, new HashMap<>()); + } + + // Extract appended file name from the absolute paths saved in getWrittenLogFiles() + pm.getWrittenLogFiles().forEach((path, size) -> { + partitionToAppendedFiles.get(partition).merge(new Path(path).getName(), size, (oldSize, newSizeCopy) -> { + return size + oldSize; + }); + }); + } }); } diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java index b79ffbbb529d9..4c28780a28e9c 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java @@ -241,7 +241,8 @@ private HoodieRollbackMetadata getRollbackMetadataInstance(String basePath, Stri List rollbacks = new ArrayList<>(); rollbacks.add(new HoodieInstant(false, actionType, commitTs)); - HoodieRollbackStat rollbackStat = new HoodieRollbackStat(partition, deletedFiles, Collections.emptyList(), Collections.emptyMap()); + HoodieRollbackStat rollbackStat = new HoodieRollbackStat(partition, deletedFiles, Collections.emptyList(), Collections.emptyMap(), + Collections.EMPTY_MAP); List rollbackStats = new ArrayList<>(); rollbackStats.add(rollbackStat); return TimelineMetadataUtils.convertRollbackMetadata(commitTs, Option.empty(), rollbacks, rollbackStats); From 6244bc69391e4ec65165c64f70f741f15b073613 Mon Sep 17 00:00:00 2001 From: Sivabalan Narayanan Date: Fri, 8 Jan 2021 11:34:54 -0500 Subject: [PATCH 3/9] Adding restore support --- .../table/action/rollback/RollbackUtils.java | 1 - .../rollback/ListingBasedRollbackHelper.java | 3 -- .../metadata/TestHoodieBackedMetadata.java | 7 ++-- .../src/main/avro/HoodieRollbackMetadata.avsc | 4 +- .../table/timeline/TimelineMetadataUtils.java | 38 +++++++++---------- .../metadata/HoodieTableMetadataUtil.java | 36 +++++++----------- .../hudi/common/table/TestTimelineUtils.java | 25 ++++++------ 7 files changed, 50 insertions(+), 64 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java index 96c99c7c89f92..176129721a816 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java @@ -123,7 +123,6 @@ public static List generateRollbackRequestsUsingFil HoodieActiveTimeline activeTimeline = table.getMetaClient().reloadActiveTimeline(); List partitionRollbackRequests = new ArrayList<>(); switch (instantToRollback.getAction()) { - case HoodieTimeline.COMMIT_ACTION: LOG.info("Rolling back commit action."); partitionRollbackRequests.add( diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java index 5c048c4d436bb..6412b450145df 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java @@ -73,7 +73,6 @@ public ListingBasedRollbackHelper(HoodieTableMetaClient metaClient, HoodieWriteC public List performRollback(HoodieEngineContext context, HoodieInstant instantToRollback, List rollbackRequests) { int sparkPartitions = Math.max(Math.min(rollbackRequests.size(), config.getRollbackParallelism()), 1); context.setJobStatus(this.getClass().getSimpleName(), "Perform rollback actions"); - JavaPairRDD partitionPathRollbackStatsPairRDD = maybeDeleteAndCollectStats(context, instantToRollback, rollbackRequests, sparkPartitions, true); return partitionPathRollbackStatsPairRDD.reduceByKey(RollbackUtils::mergeRollbackStat).map(Tuple2::_2).collect(); } @@ -149,7 +148,6 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC // if update belongs to an existing log file writer.appendBlock(new HoodieCommandBlock(header)); } - } catch (IOException | InterruptedException io) { throw new HoodieRollbackException("Failed to rollback for instant " + instantToRollback, io); } finally { @@ -169,7 +167,6 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC metaClient.getFs().getFileStatus(Objects.requireNonNull(writer).getLogFile().getPath()), 1L ); - return new Tuple2<>(rollbackRequest.getPartitionPath(), HoodieRollbackStat.newBuilder().withPartitionPath(rollbackRequest.getPartitionPath()) .withRollbackBlockAppendResults(filesToNumBlocksRollback) diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java index ed9e85b9dc875..1429e5f9b65de 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java @@ -370,9 +370,10 @@ public void testRollbackOperations(HoodieTableType tableType) throws Exception { /** * Test when syncing rollback to metadata if the commit being rolled back has not been synced that essentially a no-op occurs to metadata. */ - @Test - public void testRollbackUnsyncedCommit() throws Exception { - init(HoodieTableType.COPY_ON_WRITE); + @ParameterizedTest + @EnumSource(HoodieTableType.class) + public void testRollbackUnsyncedCommit(HoodieTableType tableType) throws Exception { + init(tableType); HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc); try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext, getWriteConfig(true, true))) { diff --git a/hudi-common/src/main/avro/HoodieRollbackMetadata.avsc b/hudi-common/src/main/avro/HoodieRollbackMetadata.avsc index e354d5ccf1efc..0132a46235c53 100644 --- a/hudi-common/src/main/avro/HoodieRollbackMetadata.avsc +++ b/hudi-common/src/main/avro/HoodieRollbackMetadata.avsc @@ -33,7 +33,6 @@ {"name": "failedDeleteFiles", "type": {"type": "array", "items": "string"}}, {"name": "rollbackLogFiles", "type": { "type": "map", - "doc": "Files to which rollback command blocks were written", "values": { "type": "long", "doc": "Size of this file in bytes" @@ -41,8 +40,7 @@ }}, {"name": "writtenLogFiles", "type": { "type": "map", - "doc": "All log files written to by the rolled back commits", - "values": { + "values": { "type": "long", "doc": "Size of this file in bytes" } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java index c29cf6829db5e..a0a4b22015e2f 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java @@ -18,17 +18,6 @@ package org.apache.hudi.common.table.timeline; -import org.apache.avro.file.DataFileReader; -import org.apache.avro.file.DataFileWriter; -import org.apache.avro.file.FileReader; -import org.apache.avro.file.SeekableByteArrayInput; -import org.apache.avro.io.DatumReader; -import org.apache.avro.io.DatumWriter; -import org.apache.avro.specific.SpecificDatumReader; -import org.apache.avro.specific.SpecificDatumWriter; -import org.apache.avro.specific.SpecificRecordBase; -import org.apache.hadoop.fs.FileStatus; - import org.apache.hudi.avro.model.HoodieCleanMetadata; import org.apache.hudi.avro.model.HoodieCleanerPlan; import org.apache.hudi.avro.model.HoodieCompactionPlan; @@ -43,6 +32,17 @@ import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.ValidationUtils; +import org.apache.avro.file.DataFileReader; +import org.apache.avro.file.DataFileWriter; +import org.apache.avro.file.FileReader; +import org.apache.avro.file.SeekableByteArrayInput; +import org.apache.avro.io.DatumReader; +import org.apache.avro.io.DatumWriter; +import org.apache.avro.specific.SpecificDatumReader; +import org.apache.avro.specific.SpecificDatumWriter; +import org.apache.avro.specific.SpecificRecordBase; +import org.apache.hadoop.fs.FileStatus; + import java.io.ByteArrayOutputStream; import java.io.IOException; import java.util.Collections; @@ -56,9 +56,9 @@ public class TimelineMetadataUtils { private static final Integer DEFAULT_VERSION = 1; public static HoodieRestoreMetadata convertRestoreMetadata(String startRestoreTime, - long durationInMs, - List instants, - Map> instantToRollbackMetadata) { + long durationInMs, + List instants, + Map> instantToRollbackMetadata) { return new HoodieRestoreMetadata(startRestoreTime, durationInMs, instants.stream().map(HoodieInstant::getTimestamp).collect(Collectors.toList()), Collections.unmodifiableMap(instantToRollbackMetadata), DEFAULT_VERSION, @@ -73,7 +73,7 @@ public static HoodieRollbackMetadata convertRollbackMetadata(String startRollbac Map rollbackLogFiles = stat.getCommandBlocksCount().keySet().stream() .collect(Collectors.toMap(f -> f.getPath().toString(), FileStatus::getLen)); Map probableLogFiles = stat.getProbableLogFileToSizeMap().keySet().stream() - .collect(Collectors.toMap(f -> f.getPath().toString(), FileStatus::getLen)); + .collect(Collectors.toMap(f -> f.getPath().toString(), FileStatus::getLen)); HoodieRollbackPartitionMetadata metadata = new HoodieRollbackPartitionMetadata(stat.getPartitionPath(), stat.getSuccessDeleteFiles(), stat.getFailedDeleteFiles(), rollbackLogFiles, probableLogFiles); partitionMetadataBuilder.put(stat.getPartitionPath(), metadata); @@ -81,9 +81,9 @@ public static HoodieRollbackMetadata convertRollbackMetadata(String startRollbac } return new HoodieRollbackMetadata(startRollbackTime, durationInMs.orElseGet(() -> -1L), totalDeleted, - instants.stream().map(HoodieInstant::getTimestamp).collect(Collectors.toList()), - Collections.unmodifiableMap(partitionMetadataBuilder), DEFAULT_VERSION, - instants.stream().map(instant -> new HoodieInstantInfo(instant.getTimestamp(), instant.getAction())).collect(Collectors.toList())); + instants.stream().map(HoodieInstant::getTimestamp).collect(Collectors.toList()), + Collections.unmodifiableMap(partitionMetadataBuilder), DEFAULT_VERSION, + instants.stream().map(instant -> new HoodieInstantInfo(instant.getTimestamp(), instant.getAction())).collect(Collectors.toList())); } public static HoodieSavepointMetadata convertSavepointMetadata(String user, String comment, @@ -94,7 +94,7 @@ public static HoodieSavepointMetadata convertSavepointMetadata(String user, Stri partitionMetadataBuilder.put(stat.getKey(), metadata); } return new HoodieSavepointMetadata(user, System.currentTimeMillis(), comment, - Collections.unmodifiableMap(partitionMetadataBuilder), DEFAULT_VERSION); + Collections.unmodifiableMap(partitionMetadataBuilder), DEFAULT_VERSION); } public static Option serializeCompactionPlan(HoodieCompactionPlan compactionWorkload) throws IOException { 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 a797485521f6d..05699b273dcd6 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,7 +18,6 @@ package org.apache.hudi.metadata; -import org.apache.hadoop.fs.Path; import org.apache.hudi.avro.model.HoodieCleanMetadata; import org.apache.hudi.avro.model.HoodieCleanerPlan; import org.apache.hudi.avro.model.HoodieRestoreMetadata; @@ -33,11 +32,14 @@ import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.ValidationUtils; import org.apache.hudi.exception.HoodieException; + +import org.apache.hadoop.fs.Path; import org.apache.log4j.LogManager; import org.apache.log4j.Logger; import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.LinkedList; @@ -60,7 +62,6 @@ public class HoodieTableMetadataUtil { * @param datasetMetaClient The meta client associated with the timeline instant * @param instant to fetch and convert to metadata table records * @return a list of metadata table records - * @throws IOException */ public static Option> convertInstantToMetaRecords(HoodieTableMetaClient datasetMetaClient, HoodieInstant instant, Option lastSyncTs) throws IOException { HoodieTimeline timeline = datasetMetaClient.getActiveTimeline(); @@ -102,8 +103,6 @@ public static Option> convertInstantToMetaRecords(HoodieTable /** * Finds all new files/partitions created as part of commit and creates metadata table records for them. * - * @param commitMetadata - * @param instantTime * @return a list of metadata table records */ public static List convertMetadataToRecords(HoodieCommitMetadata commitMetadata, String instantTime) { @@ -147,7 +146,6 @@ public static List convertMetadataToRecords(HoodieCommitMetadata c * Finds all files that will be deleted as part of a planned clean and creates metadata table records for them. * * @param cleanerPlan from timeline to convert - * @param instantTime * @return a list of metadata table records */ public static List convertMetadataToRecords(HoodieCleanerPlan cleanerPlan, String instantTime) { @@ -173,8 +171,6 @@ public static List convertMetadataToRecords(HoodieCleanerPlan clea /** * Finds all files that were deleted as part of a clean and creates metadata table records for them. * - * @param cleanMetadata - * @param instantTime * @return a list of metadata table records */ public static List convertMetadataToRecords(HoodieCleanMetadata cleanMetadata, String instantTime) { @@ -197,11 +193,8 @@ public static List convertMetadataToRecords(HoodieCleanMetadata cl } /** - * Aggregates all files deleted and appended to from all rollbacks associated with a restore operation then - * creates metadata table records for them. + * Aggregates all files deleted and appended to from all rollbacks associated with a restore operation then creates metadata table records for them. * - * @param restoreMetadata - * @param instantTime * @return a list of metadata table records */ public static List convertMetadataToRecords(HoodieRestoreMetadata restoreMetadata, String instantTime, Option lastSyncTs) { @@ -210,7 +203,6 @@ public static List convertMetadataToRecords(HoodieRestoreMetadata restoreMetadata.getHoodieRestoreMetadata().values().forEach(rms -> { rms.forEach(rm -> processRollbackMetadata(rm, partitionToDeletedFiles, partitionToAppendedFiles, lastSyncTs)); }); - return convertFilesToRecords(partitionToDeletedFiles, partitionToAppendedFiles, instantTime, "Restore"); } @@ -225,25 +217,23 @@ public static List convertMetadataToRecords(HoodieRollbackMetadata /** * Extracts information about the deleted and append files from the {@code HoodieRollbackMetadata}. * - * During a rollback files may be deleted (COW, MOR) or rollback blocks be appended (MOR only) to files. This - * function will extract this change file for each partition. + * During a rollback files may be deleted (COW, MOR) or rollback blocks be appended (MOR only) to files. This function will extract this change file for each partition. * * @param rollbackMetadata {@code HoodieRollbackMetadata} * @param partitionToDeletedFiles The {@code Map} to fill with files deleted per partition. * @param partitionToAppendedFiles The {@code Map} to fill with files appended per partition and their sizes. */ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetadata, - Map> partitionToDeletedFiles, - Map> partitionToAppendedFiles, - Option lastSyncTs) { + Map> partitionToDeletedFiles, + Map> partitionToAppendedFiles, + Option lastSyncTs) { rollbackMetadata.getPartitionMetadata().values().forEach(pm -> { // Has this rollback produced new files? - boolean hasAppendFiles = pm.getRollbackLogFiles().values().stream().mapToLong(Long::longValue).sum() > 0; + boolean hasAppendFiles = pm.getRollbackLogFiles() != null ? pm.getRollbackLogFiles().values().stream().mapToLong(Long::longValue).sum() > 0 : false; // If commit being rolled back has not been synced to metadata table yet then there is no need to update metadata boolean shouldSkip = lastSyncTs.isPresent() && HoodieTimeline.compareTimestamps(rollbackMetadata.getCommitsRollback().get(0), HoodieTimeline.GREATER_THAN, lastSyncTs.get()); - if (!hasAppendFiles && shouldSkip) { LOG.info(String.format("Skipping syncing of rollbackMetadata at %s, given metadata table is already synced upto to %s", rollbackMetadata.getCommitsRollback().get(0), lastSyncTs.get())); @@ -262,7 +252,7 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad partitionToDeletedFiles.get(partition).addAll(deletedFiles); } - if (!pm.getRollbackLogFiles().isEmpty()) { + if (pm.getRollbackLogFiles() != null && !pm.getRollbackLogFiles().isEmpty()) { if (!partitionToAppendedFiles.containsKey(partition)) { partitionToAppendedFiles.put(partition, new HashMap<>()); } @@ -275,7 +265,7 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad }); } - if (!pm.getWrittenLogFiles().isEmpty()) { + if (pm.getWrittenLogFiles() != null && !pm.getWrittenLogFiles().isEmpty()) { if (!partitionToAppendedFiles.containsKey(partition)) { partitionToAppendedFiles.put(partition, new HashMap<>()); } @@ -291,8 +281,8 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad } private static List convertFilesToRecords(Map> partitionToDeletedFiles, - Map> partitionToAppendedFiles, String instantTime, - String operation) { + Map> partitionToAppendedFiles, String instantTime, + String operation) { List records = new LinkedList<>(); int[] fileChangeCount = {0, 0}; // deletes, appends diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java index 4c28780a28e9c..32eb727de10e1 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java @@ -36,6 +36,7 @@ import org.apache.hudi.common.testutils.HoodieCommonTestHarness; import org.apache.hudi.common.util.CollectionUtils; import org.apache.hudi.common.util.Option; + import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -72,8 +73,8 @@ public void testGetPartitionsWithReplaceCommits() throws IOException { HoodieInstant instant1 = new HoodieInstant(true, HoodieTimeline.REPLACE_COMMIT_ACTION, ts1); activeTimeline.createNewInstant(instant1); // create replace metadata only with replaced file Ids (no new files created) - activeTimeline.saveAsComplete(instant1, - Option.of(getReplaceCommitMetadata(basePath, ts1, replacePartition,2, newFilePartition,0, Collections.emptyMap()))); + activeTimeline.saveAsComplete(instant1, + Option.of(getReplaceCommitMetadata(basePath, ts1, replacePartition, 2, newFilePartition, 0, Collections.emptyMap()))); metaClient.reloadActiveTimeline(); List partitions = TimelineUtils.getAffectedPartitions(metaClient.getActiveTimeline().findInstantsAfter("0", 10)); @@ -85,7 +86,7 @@ public void testGetPartitionsWithReplaceCommits() throws IOException { activeTimeline.createNewInstant(instant2); // create replace metadata only with replaced file Ids (no new files created) activeTimeline.saveAsComplete(instant2, - Option.of(getReplaceCommitMetadata(basePath, ts2, replacePartition,0, newFilePartition,3, Collections.emptyMap()))); + Option.of(getReplaceCommitMetadata(basePath, ts2, replacePartition, 0, newFilePartition, 3, Collections.emptyMap()))); metaClient.reloadActiveTimeline(); partitions = TimelineUtils.getAffectedPartitions(metaClient.getActiveTimeline().findInstantsAfter("1", 10)); assertEquals(1, partitions.size()); @@ -96,7 +97,7 @@ public void testGetPartitionsWithReplaceCommits() throws IOException { assertTrue(partitions.contains(replacePartition)); assertTrue(partitions.contains(newFilePartition)); } - + @Test public void testGetPartitions() throws IOException { HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline(); @@ -120,20 +121,20 @@ public void testGetPartitions() throws IOException { // verify modified partitions included cleaned data List partitions = TimelineUtils.getAffectedPartitions(metaClient.getActiveTimeline().findInstantsAfter("1", 10)); assertEquals(5, partitions.size()); - assertEquals(partitions, Arrays.asList(new String[]{"0", "2", "3", "4", "5"})); + assertEquals(partitions, Arrays.asList(new String[] {"0", "2", "3", "4", "5"})); partitions = TimelineUtils.getAffectedPartitions(metaClient.getActiveTimeline().findInstantsInRange("1", "4")); assertEquals(4, partitions.size()); - assertEquals(partitions, Arrays.asList(new String[]{"0", "2", "3", "4"})); + assertEquals(partitions, Arrays.asList(new String[] {"0", "2", "3", "4"})); // verify only commit actions partitions = TimelineUtils.getPartitionsWritten(metaClient.getActiveTimeline().findInstantsAfter("1", 10)); assertEquals(4, partitions.size()); - assertEquals(partitions, Arrays.asList(new String[]{"2", "3", "4", "5"})); + assertEquals(partitions, Arrays.asList(new String[] {"2", "3", "4", "5"})); partitions = TimelineUtils.getPartitionsWritten(metaClient.getActiveTimeline().findInstantsInRange("1", "4")); assertEquals(3, partitions.size()); - assertEquals(partitions, Arrays.asList(new String[]{"2", "3", "4"})); + assertEquals(partitions, Arrays.asList(new String[] {"2", "3", "4"})); } @Test @@ -181,10 +182,10 @@ public void testRestoreInstants() throws Exception { // verify modified partitions included cleaned data List partitions = TimelineUtils.getAffectedPartitions(metaClient.getActiveTimeline().findInstantsAfter("1", 10)); - assertEquals(partitions, Arrays.asList(new String[]{"2", "3", "4", "5"})); + assertEquals(partitions, Arrays.asList(new String[] {"2", "3", "4", "5"})); partitions = TimelineUtils.getAffectedPartitions(metaClient.getActiveTimeline().findInstantsInRange("1", "4")); - assertEquals(partitions, Arrays.asList(new String[]{"2", "3", "4"})); + assertEquals(partitions, Arrays.asList(new String[] {"2", "3", "4"})); } @Test @@ -201,7 +202,7 @@ public void testGetExtraMetadata() throws Exception { activeTimeline.createNewInstant(instant); activeTimeline.saveAsComplete(instant, Option.of(getCommitMetadata(basePath, ts, ts, 2, Collections.emptyMap()))); - ts = "1"; + ts = "1"; instant = new HoodieInstant(true, HoodieTimeline.COMMIT_ACTION, ts); activeTimeline.createNewInstant(instant); Map extraMetadata = new HashMap<>(); @@ -265,7 +266,7 @@ private byte[] getCommitMetadata(String basePath, String partition, String commi } private byte[] getReplaceCommitMetadata(String basePath, String commitTs, String replacePartition, int replaceCount, - String newFilePartition, int newFileCount, Map extraMetadata) + String newFilePartition, int newFileCount, Map extraMetadata) throws IOException { HoodieReplaceCommitMetadata commit = new HoodieReplaceCommitMetadata(); for (int i = 1; i <= newFileCount; i++) { From 0227058e3f34f20d9f187202b4d11f4b70dddb29 Mon Sep 17 00:00:00 2001 From: Sivabalan Narayanan Date: Fri, 8 Jan 2021 20:02:14 -0500 Subject: [PATCH 4/9] Minor fixes --- .../rollback/AbstractMarkerBasedRollbackStrategy.java | 1 - .../action/rollback/ListingBasedRollbackHelper.java | 1 - .../apache/hudi/metadata/TestHoodieBackedMetadata.java | 5 ++--- .../java/org/apache/hudi/common/HoodieRollbackStat.java | 9 --------- .../apache/hudi/metadata/HoodieTableMetadataUtil.java | 1 - .../org/apache/hudi/common/table/TestTimelineUtils.java | 1 - .../common/table/view/TestIncrementalFSViewSync.java | 2 +- 7 files changed, 3 insertions(+), 17 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java index fb6df5a227bbe..cc76734b44574 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java @@ -32,7 +32,6 @@ import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.table.HoodieTable; - import org.apache.log4j.LogManager; import org.apache.log4j.Logger; diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java index 6412b450145df..8ce5b39b05a12 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java @@ -100,7 +100,6 @@ public List collectRollbackStats(HoodieEngineContext context JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineContext context, HoodieInstant instantToRollback, List rollbackRequests, int sparkPartitions, boolean doDelete) { JavaSparkContext jsc = HoodieSparkEngineContext.getSparkContext(context); - return jsc.parallelize(rollbackRequests, sparkPartitions).mapToPair(rollbackRequest -> { switch (rollbackRequest.getType()) { case DELETE_DATA_FILES_ONLY: { diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java index 1429e5f9b65de..ce2a6aafa8f33 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java @@ -78,7 +78,6 @@ import static org.junit.jupiter.api.Assertions.assertTrue; public class TestHoodieBackedMetadata extends HoodieClientTestHarness { - private static final Logger LOG = LogManager.getLogger(TestHoodieBackedMetadata.class); @TempDir @@ -792,12 +791,12 @@ private void validateMetadata(SparkRDDWriteClient client) throws IOException { for (String fileName : fsFileNames) { if (!metadataFilenames.contains(fileName)) { - System.out.println(partition + "FsFilename " + fileName + " not found in Meta data"); + LOG.error(partition + "FsFilename " + fileName + " not found in Meta data"); } } for (String fileName : metadataFilenames) { if (!fsFileNames.contains(fileName)) { - System.out.println(partition + "Metadata file " + fileName + " not found in original FS"); + LOG.error(partition + "Metadata file " + fileName + " not found in original FS"); } } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java b/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java index 4951ba63946e9..6ba4822907f95 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java @@ -41,15 +41,6 @@ public class HoodieRollbackStat implements Serializable { // all log files with same base instant as instant to be rolledback private final Map probableLogFileToSizeMap; - public HoodieRollbackStat(String partitionPath, List successDeleteFiles, List failedDeleteFiles, - Map commandBlocksCount) { - this.partitionPath = partitionPath; - this.successDeleteFiles = successDeleteFiles; - this.failedDeleteFiles = failedDeleteFiles; - this.commandBlocksCount = commandBlocksCount; - this.probableLogFileToSizeMap = Collections.EMPTY_MAP; - } - public HoodieRollbackStat(String partitionPath, List successDeleteFiles, List failedDeleteFiles, Map commandBlocksCount, Map probableLogFileToSizeMap) { this.partitionPath = partitionPath; 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 05699b273dcd6..df001c2c060eb 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 @@ -39,7 +39,6 @@ import java.io.IOException; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.LinkedList; diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java index 32eb727de10e1..18c0d3f20a659 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java @@ -36,7 +36,6 @@ import org.apache.hudi.common.testutils.HoodieCommonTestHarness; import org.apache.hudi.common.util.CollectionUtils; import org.apache.hudi.common.util.Option; - import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestIncrementalFSViewSync.java b/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestIncrementalFSViewSync.java index 5400dc40110af..146e0bb85b53d 100644 --- a/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestIncrementalFSViewSync.java +++ b/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestIncrementalFSViewSync.java @@ -556,7 +556,7 @@ private void performRestore(HoodieInstant instant, List files, String ro boolean isRestore) throws IOException { Map> partititonToFiles = deleteFiles(files); List rollbackStats = partititonToFiles.entrySet().stream().map(e -> - new HoodieRollbackStat(e.getKey(), e.getValue(), new ArrayList<>(), new HashMap<>()) + new HoodieRollbackStat(e.getKey(), e.getValue(), new ArrayList<>(), new HashMap<>(), new HashMap<>()) ).collect(Collectors.toList()); List rollbacks = new ArrayList<>(); From 0a198d8e29af1d05583d6e634ba411981689786f Mon Sep 17 00:00:00 2001 From: Sivabalan Narayanan Date: Sat, 9 Jan 2021 17:30:01 -0500 Subject: [PATCH 5/9] Addressing feedback --- .../apache/hudi/io/HoodieAppendHandle.java | 2 +- .../AbstractMarkerBasedRollbackStrategy.java | 13 ++++---- .../table/action/rollback/RollbackUtils.java | 8 ++--- .../rollback/ListingBasedRollbackHelper.java | 26 +++++----------- .../SparkMarkerBasedRollbackStrategy.java | 26 +++++----------- .../metadata/TestHoodieBackedMetadata.java | 1 + .../TestMarkerBasedRollbackStrategy.java | 16 ++++++++-- .../hudi/common/HoodieRollbackStat.java | 22 ++++++------- .../org/apache/hudi/common/fs/FSUtils.java | 11 +++++++ .../table/timeline/TimelineMetadataUtils.java | 16 +++++----- .../metadata/HoodieTableMetadataUtil.java | 31 +++++++++++++------ 11 files changed, 94 insertions(+), 78 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java index 8f272b2c5ffc4..c6ea7bab2ca7f 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieAppendHandle.java @@ -164,7 +164,7 @@ private void init(HoodieRecord record) { // Since the actual log file written to can be different based on when rollover happens, we use the // base file to denote some log appends happened on a slice. writeToken will still fence concurrent // writers. - // TODO? are these sufficient? + // https://issues.apache.org/jira/browse/HUDI-1517 createMarkerFile(partitionPath, FSUtils.makeDataFileName(baseInstantTime, writeToken, fileId, hoodieTable.getBaseFileExtension())); this.writer = createLogWriter(fileSlice, baseInstantTime); diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java index cc76734b44574..d8446a3f8a335 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java @@ -90,7 +90,7 @@ protected HoodieRollbackStat undoAppend(String appendBaseFilePath, HoodieInstant String fileId = FSUtils.getFileIdFromFilePath(baseFilePathForAppend); String baseCommitTime = FSUtils.getCommitTime(baseFilePathForAppend.getName()); String partitionPath = FSUtils.getRelativePartitionPath(new Path(basePath), new Path(basePath, appendBaseFilePath).getParent()); - final Map probableLogFileMap = getProbableFileSizeMap(partitionPath, baseCommitTime); + final Map writtenLogFileSizeMap = getWrittenLogFileSizeMap(partitionPath, baseCommitTime, fileId); HoodieLogFormat.Writer writer = null; try { @@ -133,20 +133,21 @@ protected HoodieRollbackStat undoAppend(String appendBaseFilePath, HoodieInstant HoodieRollbackStat.Builder builder = HoodieRollbackStat.newBuilder() .withPartitionPath(partitionPath) .withRollbackBlockAppendResults(filesToNumBlocksRollback); - if (probableLogFileMap != null) { - builder.withProbableLogFileToSizeMap(probableLogFileMap); + if (!writtenLogFileSizeMap.isEmpty()) { + builder.withWrittenLogFileSizeMap(writtenLogFileSizeMap); } return builder.build(); } /** - * Returns probable log files for the respective baseCommitTime to assist in metadata table syncing. + * Returns written log file size map for the respective baseCommitTime to assist in metadata table syncing. * @param partitionPath partition path of interest * @param baseCommitTime base commit time of interest + * @param fileId fileId of interest * @return Map * @throws IOException */ - protected Map getProbableFileSizeMap(String partitionPath, String baseCommitTime) throws IOException { - return null; + protected Map getWrittenLogFileSizeMap(String partitionPath, String baseCommitTime, String fileId) throws IOException { + return Collections.EMPTY_MAP; } } diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java index 176129721a816..c2418b0f0d6af 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java @@ -74,16 +74,16 @@ static HoodieRollbackStat mergeRollbackStat(HoodieRollbackStat stat1, HoodieRoll final List successDeleteFiles = new ArrayList<>(); final List failedDeleteFiles = new ArrayList<>(); final Map commandBlocksCount = new HashMap<>(); - final Map probableLogFileMap = new HashMap<>(); + final Map writtenLogFileSizeMap = new HashMap<>(); Option.ofNullable(stat1.getSuccessDeleteFiles()).ifPresent(successDeleteFiles::addAll); Option.ofNullable(stat2.getSuccessDeleteFiles()).ifPresent(successDeleteFiles::addAll); Option.ofNullable(stat1.getFailedDeleteFiles()).ifPresent(failedDeleteFiles::addAll); Option.ofNullable(stat2.getFailedDeleteFiles()).ifPresent(failedDeleteFiles::addAll); Option.ofNullable(stat1.getCommandBlocksCount()).ifPresent(commandBlocksCount::putAll); Option.ofNullable(stat2.getCommandBlocksCount()).ifPresent(commandBlocksCount::putAll); - Option.ofNullable(stat1.getProbableLogFileToSizeMap()).ifPresent(probableLogFileMap::putAll); - Option.ofNullable(stat2.getProbableLogFileToSizeMap()).ifPresent(probableLogFileMap::putAll); - return new HoodieRollbackStat(stat1.getPartitionPath(), successDeleteFiles, failedDeleteFiles, commandBlocksCount, probableLogFileMap); + Option.ofNullable(stat1.getWrittenLogFileSizeMap()).ifPresent(writtenLogFileSizeMap::putAll); + Option.ofNullable(stat2.getWrittenLogFileSizeMap()).ifPresent(writtenLogFileSizeMap::putAll); + return new HoodieRollbackStat(stat1.getPartitionPath(), successDeleteFiles, failedDeleteFiles, commandBlocksCount, writtenLogFileSizeMap); } /** diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java index 8ce5b39b05a12..c9db9bb340f61 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java @@ -22,6 +22,7 @@ import org.apache.hudi.common.HoodieRollbackStat; import org.apache.hudi.common.engine.HoodieEngineContext; import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieFileFormat; import org.apache.hudi.common.model.HoodieLogFile; import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.common.table.log.HoodieLogFormat; @@ -49,6 +50,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.stream.Collectors; import scala.Tuple2; @@ -98,7 +100,7 @@ public List collectRollbackStats(HoodieEngineContext context * @return stats collected with or w/o actual deletions. */ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineContext context, HoodieInstant instantToRollback, List rollbackRequests, - int sparkPartitions, boolean doDelete) { + int sparkPartitions, boolean doDelete) { JavaSparkContext jsc = HoodieSparkEngineContext.getSparkContext(context); return jsc.parallelize(rollbackRequests, sparkPartitions).mapToPair(rollbackRequest -> { switch (rollbackRequest.getType()) { @@ -117,22 +119,9 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC } case APPEND_ROLLBACK_BLOCK: { // collect all log files that is supposed to be deleted with this rollback - String baseCommit = rollbackRequest.getLatestBaseInstant().get(); - SerializablePathFilter filter = (path) -> { - if (FSUtils.isLogFile(path)) { - // Since the baseCommitTime is the only commit for new log files, it's okay here - String fileCommitTime = FSUtils.getBaseCommitTimeFromLogPath(path); - return baseCommit.equals(fileCommitTime); - } - return false; - }; - - final Map probableLogFileMap = new HashMap<>(); - FileSystem fs = metaClient.getFs(); - FileStatus[] probableLogFiles = fs.listStatus(FSUtils.getPartitionPath(config.getBasePath(), rollbackRequest.getPartitionPath()), filter); - for (FileStatus fileStatus : probableLogFiles) { - probableLogFileMap.put(fileStatus, fileStatus.getLen()); - } + final Map writtenLogFileSizeMap = FSUtils.getAllLogFiles(metaClient.getFs(), FSUtils.getPartitionPath(config.getBasePath(), rollbackRequest.getPartitionPath()), + HoodieFileFormat.HOODIE_LOG.getFileExtension(), rollbackRequest.getLatestBaseInstant().get()).collect(Collectors.toMap(HoodieLogFile::getFileStatus, + value -> value.getFileStatus().getLen())); Writer writer = null; try { @@ -141,6 +130,7 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC .withFileId(rollbackRequest.getFileId().get()) .overBaseCommit(rollbackRequest.getLatestBaseInstant().get()).withFs(metaClient.getFs()) .withFileExtension(HoodieLogFile.DELTA_EXTENSION).build(); + // generate metadata if (doDelete) { Map header = generateHeader(instantToRollback.getTimestamp()); @@ -169,7 +159,7 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC return new Tuple2<>(rollbackRequest.getPartitionPath(), HoodieRollbackStat.newBuilder().withPartitionPath(rollbackRequest.getPartitionPath()) .withRollbackBlockAppendResults(filesToNumBlocksRollback) - .withProbableLogFileToSizeMap(probableLogFileMap).build()); + .withWrittenLogFileSizeMap(writtenLogFileSizeMap).build()); } default: throw new IllegalStateException("Unknown Rollback action " + rollbackRequest); diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java index d7608fd2c8e27..ef045f5df07df 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java @@ -23,7 +23,9 @@ import org.apache.hudi.common.HoodieRollbackStat; import org.apache.hudi.common.engine.HoodieEngineContext; import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieFileFormat; import org.apache.hudi.common.model.HoodieKey; +import org.apache.hudi.common.model.HoodieLogFile; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieRecordPayload; import org.apache.hudi.common.model.IOType; @@ -32,17 +34,15 @@ import org.apache.hudi.exception.HoodieRollbackException; import org.apache.hudi.table.HoodieTable; import org.apache.hudi.table.MarkerFiles; -import org.apache.hudi.table.action.rollback.ListingBasedRollbackHelper.SerializablePathFilter; import org.apache.hadoop.fs.FileStatus; -import org.apache.hadoop.fs.FileSystem; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import java.io.IOException; -import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.stream.Collectors; import scala.Tuple2; @@ -85,22 +85,10 @@ public List execute(HoodieInstant instantToRollback) { } } - protected Map getProbableFileSizeMap(String partitionPath, String baseCommitTime) throws IOException { + protected Map getWrittenLogFileSizeMap(String partitionPath, String baseCommitTime, String fileId) throws IOException { // collect all log files that is supposed to be deleted with this rollback - SerializablePathFilter filter = (path) -> { - if (FSUtils.isLogFile(path)) { - String fileCommitTime = FSUtils.getBaseCommitTimeFromLogPath(path); - return baseCommitTime.equals(fileCommitTime); - } - return false; - }; - - final Map probableLogFileMap = new HashMap<>(); - FileSystem fs = table.getMetaClient().getFs(); - FileStatus[] toBeDeleted = fs.listStatus(FSUtils.getPartitionPath(config.getBasePath(), partitionPath), filter); - for (FileStatus fileStatus : toBeDeleted) { - probableLogFileMap.put(fileStatus, fileStatus.getLen()); - } - return probableLogFileMap; + return FSUtils.getAllLogFiles(table.getMetaClient().getFs(), + FSUtils.getPartitionPath(config.getBasePath(), partitionPath), fileId, HoodieFileFormat.HOODIE_LOG.getFileExtension(), baseCommitTime) + .collect(Collectors.toMap(HoodieLogFile::getFileStatus, value -> value.getFileStatus().getLen())); } } diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java index ce2a6aafa8f33..b152b0434cfbc 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java @@ -78,6 +78,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; public class TestHoodieBackedMetadata extends HoodieClientTestHarness { + private static final Logger LOG = LogManager.getLogger(TestHoodieBackedMetadata.class); @TempDir diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMarkerBasedRollbackStrategy.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMarkerBasedRollbackStrategy.java index 7acff79ef50fb..e1e6a4c392af3 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMarkerBasedRollbackStrategy.java @@ -20,11 +20,14 @@ import org.apache.hudi.common.HoodieRollbackStat; import org.apache.hudi.common.model.HoodieFileFormat; +import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.model.IOType; import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.hudi.common.table.timeline.HoodieTimeline; import org.apache.hudi.common.testutils.HoodieTestTable; +import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.table.HoodieSparkTable; +import org.apache.hudi.table.HoodieTable; import org.apache.hudi.testutils.HoodieClientTestBase; import org.apache.hadoop.fs.FileStatus; @@ -40,12 +43,14 @@ public class TestMarkerBasedRollbackStrategy extends HoodieClientTestBase { + private HoodieTableType tableType = HoodieTableType.COPY_ON_WRITE; + @BeforeEach public void setUp() throws Exception { initPath(); initSparkContexts(); initFileSystem(); - initMetaClient(); + initMetaClient(tableType); initDFS(); } @@ -87,6 +92,11 @@ public void testCopyOnWriteRollback() throws Exception { @Test public void testMergeOnReadRollback() throws Exception { + + tearDown(); + tableType = HoodieTableType.MERGE_ON_READ; + setUp(); + // given: wrote some base + log files and corresponding markers HoodieTestTable testTable = HoodieTestTable.of(metaClient); String f2 = testTable.addRequestedDeltaCommit("000") @@ -102,8 +112,10 @@ public void testMergeOnReadRollback() throws Exception { .withMarkerFile("partA", f2, IOType.APPEND) .withMarkerFile("partB", f4, IOType.APPEND); + HoodieWriteConfig writeConfig = getConfig(); // when - List stats = new SparkMarkerBasedRollbackStrategy(HoodieSparkTable.create(getConfig(), context, metaClient), context, getConfig(), "002") + HoodieTable morTable = HoodieSparkTable.create(writeConfig, context, metaClient); + List stats = new SparkMarkerBasedRollbackStrategy(morTable, context, writeConfig, "002") .execute(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, "001")); // then: ensure files are deleted, rollback block is appended (even if append does not exist) diff --git a/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java b/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java index 6ba4822907f95..3e4ee34319c7c 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/HoodieRollbackStat.java @@ -39,15 +39,15 @@ public class HoodieRollbackStat implements Serializable { // Count of HoodieLogFile to commandBlocks written for a particular rollback private final Map commandBlocksCount; // all log files with same base instant as instant to be rolledback - private final Map probableLogFileToSizeMap; + private final Map writtenLogFileSizeMap; public HoodieRollbackStat(String partitionPath, List successDeleteFiles, List failedDeleteFiles, - Map commandBlocksCount, Map probableLogFileToSizeMap) { + Map commandBlocksCount, Map writtenLogFileSizeMap) { this.partitionPath = partitionPath; this.successDeleteFiles = successDeleteFiles; this.failedDeleteFiles = failedDeleteFiles; this.commandBlocksCount = commandBlocksCount; - this.probableLogFileToSizeMap = probableLogFileToSizeMap; + this.writtenLogFileSizeMap = writtenLogFileSizeMap; } public Map getCommandBlocksCount() { @@ -66,8 +66,8 @@ public List getFailedDeleteFiles() { return failedDeleteFiles; } - public Map getProbableLogFileToSizeMap() { - return probableLogFileToSizeMap; + public Map getWrittenLogFileSizeMap() { + return writtenLogFileSizeMap; } public static HoodieRollbackStat.Builder newBuilder() { @@ -82,7 +82,7 @@ public static class Builder { private List successDeleteFiles; private List failedDeleteFiles; private Map commandBlocksCount; - private Map probableLogFileToSizeMap; + private Map writtenLogFileSizeMap; private String partitionPath; public Builder withDeletedFileResults(Map deletedFiles) { @@ -108,8 +108,8 @@ public Builder withRollbackBlockAppendResults(Map commandBlock return this; } - public Builder withProbableLogFileToSizeMap(Map probableLogFileToSizeMap) { - this.probableLogFileToSizeMap = probableLogFileToSizeMap; + public Builder withWrittenLogFileSizeMap(Map writtenLogFileSizeMap) { + this.writtenLogFileSizeMap = writtenLogFileSizeMap; return this; } @@ -128,10 +128,10 @@ public HoodieRollbackStat build() { if (commandBlocksCount == null) { commandBlocksCount = Collections.EMPTY_MAP; } - if (probableLogFileToSizeMap == null) { - probableLogFileToSizeMap = Collections.EMPTY_MAP; + if (writtenLogFileSizeMap == null) { + writtenLogFileSizeMap = Collections.EMPTY_MAP; } - return new HoodieRollbackStat(partitionPath, successDeleteFiles, failedDeleteFiles, commandBlocksCount, probableLogFileToSizeMap); + return new HoodieRollbackStat(partitionPath, successDeleteFiles, failedDeleteFiles, commandBlocksCount, writtenLogFileSizeMap); } } } diff --git a/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java index 2d638b47e8703..d9dc3b401db75 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java @@ -430,6 +430,17 @@ public static Stream getAllLogFiles(FileSystem fs, Path partition .map(HoodieLogFile::new).filter(s -> s.getBaseCommitTime().equals(baseCommitTime)); } + /** + * Get all the log files for the passed in FileId in the partition path. + */ + public static Stream getAllLogFiles(FileSystem fs, Path partitionPath, + final String logFileExtension, final String baseCommitTime) throws IOException { + return Arrays + .stream(fs.listStatus(partitionPath, + path -> path.getName().contains(logFileExtension))) + .map(HoodieLogFile::new).filter(s -> s.getBaseCommitTime().equals(baseCommitTime)); + } + /** * Get the latest log version for the fileId in the partition path. */ diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java index a0a4b22015e2f..6011df35ab7cb 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineMetadataUtils.java @@ -56,9 +56,9 @@ public class TimelineMetadataUtils { private static final Integer DEFAULT_VERSION = 1; public static HoodieRestoreMetadata convertRestoreMetadata(String startRestoreTime, - long durationInMs, - List instants, - Map> instantToRollbackMetadata) { + long durationInMs, + List instants, + Map> instantToRollbackMetadata) { return new HoodieRestoreMetadata(startRestoreTime, durationInMs, instants.stream().map(HoodieInstant::getTimestamp).collect(Collectors.toList()), Collections.unmodifiableMap(instantToRollbackMetadata), DEFAULT_VERSION, @@ -72,7 +72,7 @@ public static HoodieRollbackMetadata convertRollbackMetadata(String startRollbac for (HoodieRollbackStat stat : rollbackStats) { Map rollbackLogFiles = stat.getCommandBlocksCount().keySet().stream() .collect(Collectors.toMap(f -> f.getPath().toString(), FileStatus::getLen)); - Map probableLogFiles = stat.getProbableLogFileToSizeMap().keySet().stream() + Map probableLogFiles = stat.getWrittenLogFileSizeMap().keySet().stream() .collect(Collectors.toMap(f -> f.getPath().toString(), FileStatus::getLen)); HoodieRollbackPartitionMetadata metadata = new HoodieRollbackPartitionMetadata(stat.getPartitionPath(), stat.getSuccessDeleteFiles(), stat.getFailedDeleteFiles(), rollbackLogFiles, probableLogFiles); @@ -81,9 +81,9 @@ public static HoodieRollbackMetadata convertRollbackMetadata(String startRollbac } return new HoodieRollbackMetadata(startRollbackTime, durationInMs.orElseGet(() -> -1L), totalDeleted, - instants.stream().map(HoodieInstant::getTimestamp).collect(Collectors.toList()), - Collections.unmodifiableMap(partitionMetadataBuilder), DEFAULT_VERSION, - instants.stream().map(instant -> new HoodieInstantInfo(instant.getTimestamp(), instant.getAction())).collect(Collectors.toList())); + instants.stream().map(HoodieInstant::getTimestamp).collect(Collectors.toList()), + Collections.unmodifiableMap(partitionMetadataBuilder), DEFAULT_VERSION, + instants.stream().map(instant -> new HoodieInstantInfo(instant.getTimestamp(), instant.getAction())).collect(Collectors.toList())); } public static HoodieSavepointMetadata convertSavepointMetadata(String user, String comment, @@ -94,7 +94,7 @@ public static HoodieSavepointMetadata convertSavepointMetadata(String user, Stri partitionMetadataBuilder.put(stat.getKey(), metadata); } return new HoodieSavepointMetadata(user, System.currentTimeMillis(), comment, - Collections.unmodifiableMap(partitionMetadataBuilder), DEFAULT_VERSION); + Collections.unmodifiableMap(partitionMetadataBuilder), DEFAULT_VERSION); } public static Option serializeCompactionPlan(HoodieCompactionPlan compactionWorkload) throws IOException { 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 df001c2c060eb..cbbee62243bf6 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 @@ -61,6 +61,7 @@ public class HoodieTableMetadataUtil { * @param datasetMetaClient The meta client associated with the timeline instant * @param instant to fetch and convert to metadata table records * @return a list of metadata table records + * @throws IOException */ public static Option> convertInstantToMetaRecords(HoodieTableMetaClient datasetMetaClient, HoodieInstant instant, Option lastSyncTs) throws IOException { HoodieTimeline timeline = datasetMetaClient.getActiveTimeline(); @@ -102,6 +103,8 @@ public static Option> convertInstantToMetaRecords(HoodieTable /** * Finds all new files/partitions created as part of commit and creates metadata table records for them. * + * @param commitMetadata + * @param instantTime * @return a list of metadata table records */ public static List convertMetadataToRecords(HoodieCommitMetadata commitMetadata, String instantTime) { @@ -145,6 +148,7 @@ public static List convertMetadataToRecords(HoodieCommitMetadata c * Finds all files that will be deleted as part of a planned clean and creates metadata table records for them. * * @param cleanerPlan from timeline to convert + * @param instantTime * @return a list of metadata table records */ public static List convertMetadataToRecords(HoodieCleanerPlan cleanerPlan, String instantTime) { @@ -170,6 +174,8 @@ public static List convertMetadataToRecords(HoodieCleanerPlan clea /** * Finds all files that were deleted as part of a clean and creates metadata table records for them. * + * @param cleanMetadata + * @param instantTime * @return a list of metadata table records */ public static List convertMetadataToRecords(HoodieCleanMetadata cleanMetadata, String instantTime) { @@ -192,8 +198,11 @@ public static List convertMetadataToRecords(HoodieCleanMetadata cl } /** - * Aggregates all files deleted and appended to from all rollbacks associated with a restore operation then creates metadata table records for them. + * Aggregates all files deleted and appended to from all rollbacks associated with a restore operation then + * creates metadata table records for them. * + * @param restoreMetadata + * @param instantTime * @return a list of metadata table records */ public static List convertMetadataToRecords(HoodieRestoreMetadata restoreMetadata, String instantTime, Option lastSyncTs) { @@ -202,6 +211,7 @@ public static List convertMetadataToRecords(HoodieRestoreMetadata restoreMetadata.getHoodieRestoreMetadata().values().forEach(rms -> { rms.forEach(rm -> processRollbackMetadata(rm, partitionToDeletedFiles, partitionToAppendedFiles, lastSyncTs)); }); + return convertFilesToRecords(partitionToDeletedFiles, partitionToAppendedFiles, instantTime, "Restore"); } @@ -216,23 +226,26 @@ public static List convertMetadataToRecords(HoodieRollbackMetadata /** * Extracts information about the deleted and append files from the {@code HoodieRollbackMetadata}. * - * During a rollback files may be deleted (COW, MOR) or rollback blocks be appended (MOR only) to files. This function will extract this change file for each partition. + * During a rollback files may be deleted (COW, MOR) or rollback blocks be appended (MOR only) to files. This + * function will extract this change file for each partition. * * @param rollbackMetadata {@code HoodieRollbackMetadata} * @param partitionToDeletedFiles The {@code Map} to fill with files deleted per partition. * @param partitionToAppendedFiles The {@code Map} to fill with files appended per partition and their sizes. */ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetadata, - Map> partitionToDeletedFiles, - Map> partitionToAppendedFiles, - Option lastSyncTs) { + Map> partitionToDeletedFiles, + Map> partitionToAppendedFiles, + Option lastSyncTs) { rollbackMetadata.getPartitionMetadata().values().forEach(pm -> { // Has this rollback produced new files? - boolean hasAppendFiles = pm.getRollbackLogFiles() != null ? pm.getRollbackLogFiles().values().stream().mapToLong(Long::longValue).sum() > 0 : false; + boolean hasRollbackLogFiles = pm.getRollbackLogFiles() != null && !pm.getRollbackLogFiles().isEmpty(); + boolean hasAppendFiles = hasRollbackLogFiles ? pm.getRollbackLogFiles().values().stream().mapToLong(Long::longValue).sum() > 0 : false; // If commit being rolled back has not been synced to metadata table yet then there is no need to update metadata boolean shouldSkip = lastSyncTs.isPresent() && HoodieTimeline.compareTimestamps(rollbackMetadata.getCommitsRollback().get(0), HoodieTimeline.GREATER_THAN, lastSyncTs.get()); + if (!hasAppendFiles && shouldSkip) { LOG.info(String.format("Skipping syncing of rollbackMetadata at %s, given metadata table is already synced upto to %s", rollbackMetadata.getCommitsRollback().get(0), lastSyncTs.get())); @@ -272,7 +285,7 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad // Extract appended file name from the absolute paths saved in getWrittenLogFiles() pm.getWrittenLogFiles().forEach((path, size) -> { partitionToAppendedFiles.get(partition).merge(new Path(path).getName(), size, (oldSize, newSizeCopy) -> { - return size + oldSize; + return size; }); }); } @@ -280,8 +293,8 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad } private static List convertFilesToRecords(Map> partitionToDeletedFiles, - Map> partitionToAppendedFiles, String instantTime, - String operation) { + Map> partitionToAppendedFiles, String instantTime, + String operation) { List records = new LinkedList<>(); int[] fileChangeCount = {0, 0}; // deletes, appends From 24ea51a460fcf382f9bce860e49fae6e24921a2b Mon Sep 17 00:00:00 2001 From: Sivabalan Narayanan Date: Sun, 10 Jan 2021 01:01:39 -0500 Subject: [PATCH 6/9] Minor fixes --- .../action/rollback/ListingBasedRollbackHelper.java | 6 ++++-- .../rollback/SparkMarkerBasedRollbackStrategy.java | 11 +++++++---- .../apache/hudi/metadata/HoodieTableMetadataUtil.java | 2 +- 3 files changed, 12 insertions(+), 7 deletions(-) diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java index c9db9bb340f61..8c2115a116048 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java @@ -37,6 +37,7 @@ import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.PathFilter; import org.apache.log4j.LogManager; import org.apache.log4j.Logger; @@ -119,9 +120,10 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC } case APPEND_ROLLBACK_BLOCK: { // collect all log files that is supposed to be deleted with this rollback - final Map writtenLogFileSizeMap = FSUtils.getAllLogFiles(metaClient.getFs(), FSUtils.getPartitionPath(config.getBasePath(), rollbackRequest.getPartitionPath()), + Path partitionPath = FSUtils.getPartitionPath(config.getBasePath(), rollbackRequest.getPartitionPath()); + final Map writtenLogFileSizeMap = metaClient.getFs().exists(partitionPath) ? FSUtils.getAllLogFiles(metaClient.getFs(), partitionPath, HoodieFileFormat.HOODIE_LOG.getFileExtension(), rollbackRequest.getLatestBaseInstant().get()).collect(Collectors.toMap(HoodieLogFile::getFileStatus, - value -> value.getFileStatus().getLen())); + value -> value.getFileStatus().getLen())) : Collections.EMPTY_MAP; Writer writer = null; try { diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java index ef045f5df07df..bb13870515850 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java @@ -36,10 +36,12 @@ import org.apache.hudi.table.MarkerFiles; import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.Path; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import java.io.IOException; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.stream.Collectors; @@ -85,10 +87,11 @@ public List execute(HoodieInstant instantToRollback) { } } - protected Map getWrittenLogFileSizeMap(String partitionPath, String baseCommitTime, String fileId) throws IOException { + protected Map getWrittenLogFileSizeMap(String partitionPathStr, String baseCommitTime, String fileId) throws IOException { // collect all log files that is supposed to be deleted with this rollback - return FSUtils.getAllLogFiles(table.getMetaClient().getFs(), - FSUtils.getPartitionPath(config.getBasePath(), partitionPath), fileId, HoodieFileFormat.HOODIE_LOG.getFileExtension(), baseCommitTime) - .collect(Collectors.toMap(HoodieLogFile::getFileStatus, value -> value.getFileStatus().getLen())); + Path partitionPath = FSUtils.getPartitionPath(config.getBasePath(), partitionPathStr); + return table.getMetaClient().getFs().exists(partitionPath) ? FSUtils.getAllLogFiles(table.getMetaClient().getFs(), + FSUtils.getPartitionPath(config.getBasePath(), partitionPathStr), HoodieFileFormat.HOODIE_LOG.getFileExtension(), baseCommitTime) + .collect(Collectors.toMap(HoodieLogFile::getFileStatus, value -> value.getFileStatus().getLen())) : Collections.EMPTY_MAP; } } 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 cbbee62243bf6..a6c4fbaf6609f 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 @@ -285,7 +285,7 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad // Extract appended file name from the absolute paths saved in getWrittenLogFiles() pm.getWrittenLogFiles().forEach((path, size) -> { partitionToAppendedFiles.get(partition).merge(new Path(path).getName(), size, (oldSize, newSizeCopy) -> { - return size; + return oldSize; }); }); } From 69b4b805516bc4ec70a2595510ba8150f4ffd286 Mon Sep 17 00:00:00 2001 From: Sivabalan Narayanan Date: Sun, 10 Jan 2021 16:52:19 -0500 Subject: [PATCH 7/9] Fixing TestMarkerBasedRollbackStrategy --- .../AbstractMarkerBasedRollbackStrategy.java | 12 +- .../rollback/ListingBasedRollbackHelper.java | 11 +- .../TestMarkerBasedRollbackStrategy.java | 163 ++++++++++-------- 3 files changed, 99 insertions(+), 87 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java index d8446a3f8a335..b049b5ad38d50 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java @@ -90,7 +90,8 @@ protected HoodieRollbackStat undoAppend(String appendBaseFilePath, HoodieInstant String fileId = FSUtils.getFileIdFromFilePath(baseFilePathForAppend); String baseCommitTime = FSUtils.getCommitTime(baseFilePathForAppend.getName()); String partitionPath = FSUtils.getRelativePartitionPath(new Path(basePath), new Path(basePath, appendBaseFilePath).getParent()); - final Map writtenLogFileSizeMap = getWrittenLogFileSizeMap(partitionPath, baseCommitTime, fileId); + final Map writtenLogFileSizeMap = config.useFileListingMetadata() + ? getWrittenLogFileSizeMap(partitionPath, baseCommitTime, fileId) : Collections.EMPTY_MAP; HoodieLogFormat.Writer writer = null; try { @@ -130,13 +131,10 @@ protected HoodieRollbackStat undoAppend(String appendBaseFilePath, HoodieInstant 1L); } - HoodieRollbackStat.Builder builder = HoodieRollbackStat.newBuilder() + return HoodieRollbackStat.newBuilder() .withPartitionPath(partitionPath) - .withRollbackBlockAppendResults(filesToNumBlocksRollback); - if (!writtenLogFileSizeMap.isEmpty()) { - builder.withWrittenLogFileSizeMap(writtenLogFileSizeMap); - } - return builder.build(); + .withRollbackBlockAppendResults(filesToNumBlocksRollback) + .withWrittenLogFileSizeMap(writtenLogFileSizeMap).build(); } /** diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java index 8c2115a116048..5ee122d6b1292 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java @@ -120,10 +120,13 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC } case APPEND_ROLLBACK_BLOCK: { // collect all log files that is supposed to be deleted with this rollback - Path partitionPath = FSUtils.getPartitionPath(config.getBasePath(), rollbackRequest.getPartitionPath()); - final Map writtenLogFileSizeMap = metaClient.getFs().exists(partitionPath) ? FSUtils.getAllLogFiles(metaClient.getFs(), partitionPath, - HoodieFileFormat.HOODIE_LOG.getFileExtension(), rollbackRequest.getLatestBaseInstant().get()).collect(Collectors.toMap(HoodieLogFile::getFileStatus, - value -> value.getFileStatus().getLen())) : Collections.EMPTY_MAP; + Map writtenLogFileSizeMap = Collections.EMPTY_MAP; + if (config.useFileListingMetadata()) { + Path partitionPath = FSUtils.getPartitionPath(config.getBasePath(), rollbackRequest.getPartitionPath()); + writtenLogFileSizeMap = metaClient.getFs().exists(partitionPath) ? FSUtils.getAllLogFiles(metaClient.getFs(), partitionPath, + HoodieFileFormat.HOODIE_LOG.getFileExtension(), rollbackRequest.getLatestBaseInstant().get()).collect(Collectors.toMap(HoodieLogFile::getFileStatus, + value -> value.getFileStatus().getLen())) : Collections.EMPTY_MAP; + } Writer writer = null; try { diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMarkerBasedRollbackStrategy.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMarkerBasedRollbackStrategy.java index e1e6a4c392af3..cc3210142acf2 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMarkerBasedRollbackStrategy.java @@ -18,23 +18,28 @@ package org.apache.hudi.table.action.rollback; +import org.apache.hudi.client.SparkRDDWriteClient; +import org.apache.hudi.client.WriteStatus; +import org.apache.hudi.client.common.HoodieSparkEngineContext; import org.apache.hudi.common.HoodieRollbackStat; +import org.apache.hudi.common.config.HoodieMetadataConfig; import org.apache.hudi.common.model.HoodieFileFormat; +import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieTableType; -import org.apache.hudi.common.model.IOType; import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.hudi.common.table.timeline.HoodieTimeline; -import org.apache.hudi.common.testutils.HoodieTestTable; import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.table.HoodieSparkTable; -import org.apache.hudi.table.HoodieTable; import org.apache.hudi.testutils.HoodieClientTestBase; -import org.apache.hadoop.fs.FileStatus; +import org.apache.spark.api.java.JavaRDD; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import java.util.Arrays; import java.util.List; import java.util.stream.Stream; @@ -43,6 +48,12 @@ public class TestMarkerBasedRollbackStrategy extends HoodieClientTestBase { + private static final String TEST_NAME_WITH_PARAMS = "[{index}] Test with listing metadata enable={0}"; + + public static Stream configParams() { + return Arrays.stream(new Boolean[][] {{true}, {false}}).map(Arguments::of); + } + private HoodieTableType tableType = HoodieTableType.COPY_ON_WRITE; @BeforeEach @@ -51,7 +62,7 @@ public void setUp() throws Exception { initSparkContexts(); initFileSystem(); initMetaClient(tableType); - initDFS(); + initTestDataGenerator(); } @AfterEach @@ -59,81 +70,81 @@ public void tearDown() throws Exception { cleanupResources(); } - @Test - public void testCopyOnWriteRollback() throws Exception { - // given: wrote some base files and corresponding markers - HoodieTestTable testTable = HoodieTestTable.of(metaClient); - String f0 = testTable.addRequestedCommit("000") - .getFileIdsWithBaseFilesInPartitions("partA").get("partA"); - String f1 = testTable.addCommit("001") - .withBaseFilesInPartition("partA", f0) - .getFileIdsWithBaseFilesInPartitions("partB").get("partB"); - String f2 = "f2"; - testTable.forCommit("001") - .withMarkerFile("partA", f0, IOType.MERGE) - .withMarkerFile("partB", f1, IOType.CREATE) - .withMarkerFile("partA", f2, IOType.CREATE); - - // when - List stats = new SparkMarkerBasedRollbackStrategy(HoodieSparkTable.create(getConfig(), context, metaClient), context, getConfig(), "002") - .execute(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.COMMIT_ACTION, "001")); - - // then: ensure files are deleted correctly, non-existent files reported as failed deletes - assertEquals(2, stats.size()); - - FileStatus[] partAFiles = testTable.listAllFilesInPartition("partA"); - FileStatus[] partBFiles = testTable.listAllFilesInPartition("partB"); - - assertEquals(0, partBFiles.length); - assertEquals(1, partAFiles.length); - assertEquals(2, stats.stream().mapToInt(r -> r.getSuccessDeleteFiles().size()).sum()); - assertEquals(1, stats.stream().mapToInt(r -> r.getFailedDeleteFiles().size()).sum()); + @ParameterizedTest(name = TEST_NAME_WITH_PARAMS) + @MethodSource("configParams") + public void testCopyOnWriteRollback(boolean useFileListingMetadata) throws Exception { + HoodieWriteConfig writeConfig = getConfigBuilder().withRollbackUsingMarkers(true).withAutoCommit(false) + .withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(useFileListingMetadata).build()) + .withPath(basePath).build(); + + HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc); + try (SparkRDDWriteClient writeClient = new SparkRDDWriteClient(engineContext, writeConfig)) { + // rollback 2nd commit and ensure stats reflect the info. + List stats = testRun(useFileListingMetadata, writeConfig, writeClient); + + assertEquals(3, stats.size()); + for (HoodieRollbackStat stat : stats) { + assertEquals(1, stat.getSuccessDeleteFiles().size()); + assertEquals(0, stat.getFailedDeleteFiles().size()); + assertEquals(0, stat.getCommandBlocksCount().size()); + assertEquals(0, stat.getWrittenLogFileSizeMap().size()); + } + } } - @Test - public void testMergeOnReadRollback() throws Exception { - + @ParameterizedTest(name = TEST_NAME_WITH_PARAMS) + @MethodSource("configParams") + public void testMergeOnReadRollback(boolean useFileListingMetadata) throws Exception { + // init MERGE_ON_READ_TABLE tearDown(); tableType = HoodieTableType.MERGE_ON_READ; setUp(); - // given: wrote some base + log files and corresponding markers - HoodieTestTable testTable = HoodieTestTable.of(metaClient); - String f2 = testTable.addRequestedDeltaCommit("000") - .getFileIdsWithBaseFilesInPartitions("partA").get("partA"); - String f1 = testTable.addDeltaCommit("001") - .withLogFile("partA", f2) - .getFileIdsWithBaseFilesInPartitions("partB").get("partB"); - String f3 = "f3"; - String f4 = "f4"; - testTable.forDeltaCommit("001") - .withMarkerFile("partB", f1, IOType.CREATE) - .withMarkerFile("partA", f3, IOType.CREATE) - .withMarkerFile("partA", f2, IOType.APPEND) - .withMarkerFile("partB", f4, IOType.APPEND); - - HoodieWriteConfig writeConfig = getConfig(); - // when - HoodieTable morTable = HoodieSparkTable.create(writeConfig, context, metaClient); - List stats = new SparkMarkerBasedRollbackStrategy(morTable, context, writeConfig, "002") - .execute(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, "001")); - - // then: ensure files are deleted, rollback block is appended (even if append does not exist) - assertEquals(2, stats.size()); - // will have the log file - FileStatus[] partBFiles = testTable.listAllFilesInPartition("partB"); - assertEquals(1, partBFiles.length); - assertTrue(partBFiles[0].getPath().getName().contains(HoodieFileFormat.HOODIE_LOG.getFileExtension())); - assertTrue(partBFiles[0].getLen() > 0); - - FileStatus[] partAFiles = testTable.listAllFilesInPartition("partA"); - assertEquals(3, partAFiles.length); - assertEquals(2, Stream.of(partAFiles).filter(s -> s.getPath().getName().contains(HoodieFileFormat.HOODIE_LOG.getFileExtension())).count()); - assertEquals(1, Stream.of(partAFiles).filter(s -> s.getPath().getName().contains(HoodieFileFormat.HOODIE_LOG.getFileExtension())).filter(f -> f.getLen() > 0).count()); - - // only partB/f1_001 will be deleted - assertEquals(1, stats.stream().mapToInt(r -> r.getSuccessDeleteFiles().size()).sum()); - // partA/f3_001 is non existent - assertEquals(1, stats.stream().mapToInt(r -> r.getFailedDeleteFiles().size()).sum()); + HoodieWriteConfig writeConfig = getConfigBuilder().withRollbackUsingMarkers(true).withAutoCommit(false) + .withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(useFileListingMetadata).build()) + .withPath(basePath).build(); + + HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc); + try (SparkRDDWriteClient writeClient = new SparkRDDWriteClient(engineContext, writeConfig)) { + + // rollback 2nd commit and ensure stats reflect the info. + List stats = testRun(useFileListingMetadata, writeConfig, writeClient); + + assertEquals(3, stats.size()); + for (HoodieRollbackStat stat : stats) { + assertEquals(0, stat.getSuccessDeleteFiles().size()); + assertEquals(0, stat.getFailedDeleteFiles().size()); + if (useFileListingMetadata) { + assertEquals(1, stat.getCommandBlocksCount().size()); + stat.getCommandBlocksCount().forEach((fileStatus, len) -> assertTrue(fileStatus.getPath().getName().contains(HoodieFileFormat.HOODIE_LOG.getFileExtension()))); + assertEquals(1, stat.getWrittenLogFileSizeMap().size()); + stat.getWrittenLogFileSizeMap().forEach((fileStatus, len) -> assertTrue(fileStatus.getPath().getName().contains(HoodieFileFormat.HOODIE_LOG.getFileExtension()))); + } else { + assertEquals(0, stat.getCommandBlocksCount().size()); + assertEquals(0, stat.getWrittenLogFileSizeMap().size()); + } + } + } + } + + private List testRun(boolean useFileListingMetadata, HoodieWriteConfig writeConfig, SparkRDDWriteClient writeClient) { + String newCommitTime = "001"; + writeClient.startCommitWithTime(newCommitTime); + + List records = dataGen.generateInserts(newCommitTime, 100); + JavaRDD writeStatuses = writeClient.insert(jsc.parallelize(records, 1), newCommitTime); + writeClient.commit(newCommitTime, writeStatuses); + + // Updates + newCommitTime = "002"; + writeClient.startCommitWithTime(newCommitTime); + records = dataGen.generateUniqueUpdates(newCommitTime, 50); + writeStatuses = writeClient.upsert(jsc.parallelize(records, 1), newCommitTime); + writeStatuses.collect(); + + // rollback 2nd commit and ensure stats reflect the info. + return new SparkMarkerBasedRollbackStrategy(HoodieSparkTable.create(writeConfig, context, metaClient), context, writeConfig, "003") + .execute(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, "002")); } + } From 6df8662cbf52448855ee462f1e0f6142cc3dbe3f Mon Sep 17 00:00:00 2001 From: Sivabalan Narayanan Date: Sun, 10 Jan 2021 21:31:31 -0500 Subject: [PATCH 8/9] Addressing more comments --- .../AbstractMarkerBasedRollbackStrategy.java | 10 ++-- .../rollback/ListingBasedRollbackHelper.java | 13 ++--- .../SparkMarkerBasedRollbackStrategy.java | 13 ++--- .../metadata/TestHoodieBackedMetadata.java | 4 +- .../TestMarkerBasedRollbackStrategy.java | 51 +++++++++++++++---- .../src/main/avro/HoodieRollbackMetadata.avsc | 2 + .../org/apache/hudi/common/fs/FSUtils.java | 25 ++++++--- .../metadata/HoodieTableMetadataUtil.java | 10 ++-- 8 files changed, 82 insertions(+), 46 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java index b049b5ad38d50..cc596ba3422b7 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/AbstractMarkerBasedRollbackStrategy.java @@ -90,8 +90,7 @@ protected HoodieRollbackStat undoAppend(String appendBaseFilePath, HoodieInstant String fileId = FSUtils.getFileIdFromFilePath(baseFilePathForAppend); String baseCommitTime = FSUtils.getCommitTime(baseFilePathForAppend.getName()); String partitionPath = FSUtils.getRelativePartitionPath(new Path(basePath), new Path(basePath, appendBaseFilePath).getParent()); - final Map writtenLogFileSizeMap = config.useFileListingMetadata() - ? getWrittenLogFileSizeMap(partitionPath, baseCommitTime, fileId) : Collections.EMPTY_MAP; + final Map writtenLogFileSizeMap = getWrittenLogFileSizeMap(partitionPath, baseCommitTime, fileId); HoodieLogFormat.Writer writer = null; try { @@ -123,13 +122,10 @@ protected HoodieRollbackStat undoAppend(String appendBaseFilePath, HoodieInstant } } - Map filesToNumBlocksRollback = Collections.emptyMap(); - if (config.useFileListingMetadata()) { - // When metadata is enabled, the information of files appended to is required - filesToNumBlocksRollback = Collections.singletonMap( + // the information of files appended to is required for metadata sync + Map filesToNumBlocksRollback = Collections.singletonMap( table.getMetaClient().getFs().getFileStatus(Objects.requireNonNull(writer).getLogFile().getPath()), 1L); - } return HoodieRollbackStat.newBuilder() .withPartitionPath(partitionPath) diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java index 5ee122d6b1292..e458ead50cf24 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java @@ -37,7 +37,6 @@ import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.PathFilter; import org.apache.log4j.LogManager; import org.apache.log4j.Logger; @@ -120,13 +119,10 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC } case APPEND_ROLLBACK_BLOCK: { // collect all log files that is supposed to be deleted with this rollback - Map writtenLogFileSizeMap = Collections.EMPTY_MAP; - if (config.useFileListingMetadata()) { - Path partitionPath = FSUtils.getPartitionPath(config.getBasePath(), rollbackRequest.getPartitionPath()); - writtenLogFileSizeMap = metaClient.getFs().exists(partitionPath) ? FSUtils.getAllLogFiles(metaClient.getFs(), partitionPath, - HoodieFileFormat.HOODIE_LOG.getFileExtension(), rollbackRequest.getLatestBaseInstant().get()).collect(Collectors.toMap(HoodieLogFile::getFileStatus, - value -> value.getFileStatus().getLen())) : Collections.EMPTY_MAP; - } + Map writtenLogFileSizeMap = FSUtils.getAllLogFiles(metaClient.getFs(), + FSUtils.getPartitionPath(config.getBasePath(), rollbackRequest.getPartitionPath()), + rollbackRequest.getFileId().get(), HoodieFileFormat.HOODIE_LOG.getFileExtension(), + rollbackRequest.getLatestBaseInstant().get()).collect(Collectors.toMap(HoodieLogFile::getFileStatus, value -> value.getFileStatus().getLen())); Writer writer = null; try { @@ -161,6 +157,7 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC metaClient.getFs().getFileStatus(Objects.requireNonNull(writer).getLogFile().getPath()), 1L ); + return new Tuple2<>(rollbackRequest.getPartitionPath(), HoodieRollbackStat.newBuilder().withPartitionPath(rollbackRequest.getPartitionPath()) .withRollbackBlockAppendResults(filesToNumBlocksRollback) diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java index bb13870515850..0f36cb85aa40d 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/SparkMarkerBasedRollbackStrategy.java @@ -36,12 +36,10 @@ import org.apache.hudi.table.MarkerFiles; import org.apache.hadoop.fs.FileStatus; -import org.apache.hadoop.fs.Path; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import java.io.IOException; -import java.util.Collections; import java.util.List; import java.util.Map; import java.util.stream.Collectors; @@ -50,9 +48,7 @@ @SuppressWarnings("checkstyle:LineLength") public class SparkMarkerBasedRollbackStrategy extends AbstractMarkerBasedRollbackStrategy>, JavaRDD, JavaRDD> { - - public SparkMarkerBasedRollbackStrategy(HoodieTable>, JavaRDD, JavaRDD> table, HoodieEngineContext context, HoodieWriteConfig config, - String instantTime) { + public SparkMarkerBasedRollbackStrategy(HoodieTable>, JavaRDD, JavaRDD> table, HoodieEngineContext context, HoodieWriteConfig config, String instantTime) { super(table, context, config, instantTime); } @@ -89,9 +85,8 @@ public List execute(HoodieInstant instantToRollback) { protected Map getWrittenLogFileSizeMap(String partitionPathStr, String baseCommitTime, String fileId) throws IOException { // collect all log files that is supposed to be deleted with this rollback - Path partitionPath = FSUtils.getPartitionPath(config.getBasePath(), partitionPathStr); - return table.getMetaClient().getFs().exists(partitionPath) ? FSUtils.getAllLogFiles(table.getMetaClient().getFs(), - FSUtils.getPartitionPath(config.getBasePath(), partitionPathStr), HoodieFileFormat.HOODIE_LOG.getFileExtension(), baseCommitTime) - .collect(Collectors.toMap(HoodieLogFile::getFileStatus, value -> value.getFileStatus().getLen())) : Collections.EMPTY_MAP; + return FSUtils.getAllLogFiles(table.getMetaClient().getFs(), + FSUtils.getPartitionPath(config.getBasePath(), partitionPathStr), fileId, HoodieFileFormat.HOODIE_LOG.getFileExtension(), baseCommitTime) + .collect(Collectors.toMap(HoodieLogFile::getFileStatus, value -> value.getFileStatus().getLen())); } } diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java index b152b0434cfbc..c5fe69d42c366 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/metadata/TestHoodieBackedMetadata.java @@ -369,6 +369,7 @@ public void testRollbackOperations(HoodieTableType tableType) throws Exception { /** * Test when syncing rollback to metadata if the commit being rolled back has not been synced that essentially a no-op occurs to metadata. + * Once explicit sync is called, metadata should match. */ @ParameterizedTest @EnumSource(HoodieTableType.class) @@ -385,7 +386,6 @@ public void testRollbackUnsyncedCommit(HoodieTableType tableType) throws Excepti assertNoWriteErrors(writeStatuses); validateMetadata(client); } - String newCommitTime = HoodieActiveTimeline.createNewInstantTime(); try (SparkRDDWriteClient client = new SparkRDDWriteClient(engineContext, getWriteConfig(true, false))) { // Commit with metadata disabled @@ -397,6 +397,8 @@ public void testRollbackUnsyncedCommit(HoodieTableType tableType) throws Excepti } try (SparkRDDWriteClient client = new SparkRDDWriteClient<>(engineContext, getWriteConfig(true, true))) { + assertFalse(metadata(client).isInSync()); + client.syncTableMetadata(); validateMetadata(client); } } diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMarkerBasedRollbackStrategy.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMarkerBasedRollbackStrategy.java index cc3210142acf2..83e2b0584eed2 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMarkerBasedRollbackStrategy.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMarkerBasedRollbackStrategy.java @@ -26,15 +26,20 @@ import org.apache.hudi.common.model.HoodieFileFormat; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.model.IOType; import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.testutils.HoodieTestTable; import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.table.HoodieSparkTable; import org.apache.hudi.testutils.HoodieClientTestBase; +import org.apache.hadoop.fs.FileStatus; import org.apache.spark.api.java.JavaRDD; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; @@ -70,6 +75,38 @@ public void tearDown() throws Exception { cleanupResources(); } + @Test + public void testCopyOnWriteRollbackWithTestTable() throws Exception { + // given: wrote some base files and corresponding markers + HoodieTestTable testTable = HoodieTestTable.of(metaClient); + String f0 = testTable.addRequestedCommit("000") + .getFileIdsWithBaseFilesInPartitions("partA").get("partA"); + String f1 = testTable.addCommit("001") + .withBaseFilesInPartition("partA", f0) + .getFileIdsWithBaseFilesInPartitions("partB").get("partB"); + String f2 = "f2"; + testTable.forCommit("001") + .withMarkerFile("partA", f0, IOType.MERGE) + .withMarkerFile("partB", f1, IOType.CREATE) + .withMarkerFile("partA", f2, IOType.CREATE); + + // when + List stats = new SparkMarkerBasedRollbackStrategy(HoodieSparkTable.create(getConfig(), context, metaClient), context, getConfig(), "002") + .execute(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.COMMIT_ACTION, "001")); + + // then: ensure files are deleted correctly, non-existent files reported as failed deletes + assertEquals(2, stats.size()); + + FileStatus[] partAFiles = testTable.listAllFilesInPartition("partA"); + FileStatus[] partBFiles = testTable.listAllFilesInPartition("partB"); + + assertEquals(0, partBFiles.length); + assertEquals(1, partAFiles.length); + assertEquals(2, stats.stream().mapToInt(r -> r.getSuccessDeleteFiles().size()).sum()); + assertEquals(1, stats.stream().mapToInt(r -> r.getFailedDeleteFiles().size()).sum()); + } + + @Tag("functional") @ParameterizedTest(name = TEST_NAME_WITH_PARAMS) @MethodSource("configParams") public void testCopyOnWriteRollback(boolean useFileListingMetadata) throws Exception { @@ -92,6 +129,7 @@ public void testCopyOnWriteRollback(boolean useFileListingMetadata) throws Excep } } + @Tag("functional") @ParameterizedTest(name = TEST_NAME_WITH_PARAMS) @MethodSource("configParams") public void testMergeOnReadRollback(boolean useFileListingMetadata) throws Exception { @@ -114,15 +152,10 @@ public void testMergeOnReadRollback(boolean useFileListingMetadata) throws Excep for (HoodieRollbackStat stat : stats) { assertEquals(0, stat.getSuccessDeleteFiles().size()); assertEquals(0, stat.getFailedDeleteFiles().size()); - if (useFileListingMetadata) { - assertEquals(1, stat.getCommandBlocksCount().size()); - stat.getCommandBlocksCount().forEach((fileStatus, len) -> assertTrue(fileStatus.getPath().getName().contains(HoodieFileFormat.HOODIE_LOG.getFileExtension()))); - assertEquals(1, stat.getWrittenLogFileSizeMap().size()); - stat.getWrittenLogFileSizeMap().forEach((fileStatus, len) -> assertTrue(fileStatus.getPath().getName().contains(HoodieFileFormat.HOODIE_LOG.getFileExtension()))); - } else { - assertEquals(0, stat.getCommandBlocksCount().size()); - assertEquals(0, stat.getWrittenLogFileSizeMap().size()); - } + assertEquals(1, stat.getCommandBlocksCount().size()); + stat.getCommandBlocksCount().forEach((fileStatus, len) -> assertTrue(fileStatus.getPath().getName().contains(HoodieFileFormat.HOODIE_LOG.getFileExtension()))); + assertEquals(1, stat.getWrittenLogFileSizeMap().size()); + stat.getWrittenLogFileSizeMap().forEach((fileStatus, len) -> assertTrue(fileStatus.getPath().getName().contains(HoodieFileFormat.HOODIE_LOG.getFileExtension()))); } } } diff --git a/hudi-common/src/main/avro/HoodieRollbackMetadata.avsc b/hudi-common/src/main/avro/HoodieRollbackMetadata.avsc index 0132a46235c53..a972bfd525e2d 100644 --- a/hudi-common/src/main/avro/HoodieRollbackMetadata.avsc +++ b/hudi-common/src/main/avro/HoodieRollbackMetadata.avsc @@ -33,6 +33,7 @@ {"name": "failedDeleteFiles", "type": {"type": "array", "items": "string"}}, {"name": "rollbackLogFiles", "type": { "type": "map", + "doc": "Files to which append blocks were written to capture rollback commit", "values": { "type": "long", "doc": "Size of this file in bytes" @@ -40,6 +41,7 @@ }}, {"name": "writtenLogFiles", "type": { "type": "map", + "doc": "Log files written that were expected to be rolledback", "values": { "type": "long", "doc": "Size of this file in bytes" diff --git a/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java index d9dc3b401db75..642543d57979f 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java @@ -44,6 +44,7 @@ import org.apache.log4j.Logger; import java.io.File; +import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; @@ -424,10 +425,14 @@ public static Option getLatestLogFile(Stream logFi */ public static Stream getAllLogFiles(FileSystem fs, Path partitionPath, final String fileId, final String logFileExtension, final String baseCommitTime) throws IOException { - return Arrays - .stream(fs.listStatus(partitionPath, - path -> path.getName().startsWith("." + fileId) && path.getName().contains(logFileExtension))) - .map(HoodieLogFile::new).filter(s -> s.getBaseCommitTime().equals(baseCommitTime)); + try { + return Arrays + .stream(fs.listStatus(partitionPath, + path -> path.getName().startsWith("." + fileId) && path.getName().contains(logFileExtension))) + .map(HoodieLogFile::new).filter(s -> s.getBaseCommitTime().equals(baseCommitTime)); + } catch (FileNotFoundException e) { + return Stream.builder().build(); + } } /** @@ -435,10 +440,14 @@ public static Stream getAllLogFiles(FileSystem fs, Path partition */ public static Stream getAllLogFiles(FileSystem fs, Path partitionPath, final String logFileExtension, final String baseCommitTime) throws IOException { - return Arrays - .stream(fs.listStatus(partitionPath, - path -> path.getName().contains(logFileExtension))) - .map(HoodieLogFile::new).filter(s -> s.getBaseCommitTime().equals(baseCommitTime)); + try { + return Arrays + .stream(fs.listStatus(partitionPath, + path -> path.getName().contains(logFileExtension))) + .map(HoodieLogFile::new).filter(s -> s.getBaseCommitTime().equals(baseCommitTime)); + } catch (FileNotFoundException e) { + return Stream.builder().build(); + } } /** 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 a6c4fbaf6609f..b3d928749bae1 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 @@ -241,12 +241,12 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad rollbackMetadata.getPartitionMetadata().values().forEach(pm -> { // Has this rollback produced new files? boolean hasRollbackLogFiles = pm.getRollbackLogFiles() != null && !pm.getRollbackLogFiles().isEmpty(); - boolean hasAppendFiles = hasRollbackLogFiles ? pm.getRollbackLogFiles().values().stream().mapToLong(Long::longValue).sum() > 0 : false; + boolean hasNonZeroRollbackLogFiles = hasRollbackLogFiles ? pm.getRollbackLogFiles().values().stream().mapToLong(Long::longValue).sum() > 0 : false; // If commit being rolled back has not been synced to metadata table yet then there is no need to update metadata boolean shouldSkip = lastSyncTs.isPresent() && HoodieTimeline.compareTimestamps(rollbackMetadata.getCommitsRollback().get(0), HoodieTimeline.GREATER_THAN, lastSyncTs.get()); - if (!hasAppendFiles && shouldSkip) { + if (!hasNonZeroRollbackLogFiles && shouldSkip) { LOG.info(String.format("Skipping syncing of rollbackMetadata at %s, given metadata table is already synced upto to %s", rollbackMetadata.getCommitsRollback().get(0), lastSyncTs.get())); return; @@ -264,7 +264,7 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad partitionToDeletedFiles.get(partition).addAll(deletedFiles); } - if (pm.getRollbackLogFiles() != null && !pm.getRollbackLogFiles().isEmpty()) { + if (hasRollbackLogFiles) { if (!partitionToAppendedFiles.containsKey(partition)) { partitionToAppendedFiles.put(partition, new HashMap<>()); } @@ -285,7 +285,9 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad // Extract appended file name from the absolute paths saved in getWrittenLogFiles() pm.getWrittenLogFiles().forEach((path, size) -> { partitionToAppendedFiles.get(partition).merge(new Path(path).getName(), size, (oldSize, newSizeCopy) -> { - return oldSize; + // if a file exists in both written log files and rollback log files, we want to pick the one that is higher + // as rollback file could have been updated after written log files are computed. + return oldSize > newSizeCopy ? oldSize : newSizeCopy; }); }); } From c2647c30713c7099a369cbcf727d577ac64710a4 Mon Sep 17 00:00:00 2001 From: Vinoth Chandar Date: Mon, 11 Jan 2021 09:18:42 -0800 Subject: [PATCH 9/9] CR feedback --- .../rollback/ListingBasedRollbackHelper.java | 12 ++++++++---- .../org/apache/hudi/common/fs/FSUtils.java | 15 --------------- .../metadata/HoodieTableMetadataUtil.java | 19 ++++++++++--------- 3 files changed, 18 insertions(+), 28 deletions(-) diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java index e458ead50cf24..fcb3882b7bf5e 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java @@ -118,18 +118,22 @@ JavaPairRDD maybeDeleteAndCollectStats(HoodieEngineC .withDeletedFileResults(filesToDeletedStatus).build()); } case APPEND_ROLLBACK_BLOCK: { + String fileId = rollbackRequest.getFileId().get(); + String latestBaseInstant = rollbackRequest.getLatestBaseInstant().get(); + // collect all log files that is supposed to be deleted with this rollback Map writtenLogFileSizeMap = FSUtils.getAllLogFiles(metaClient.getFs(), FSUtils.getPartitionPath(config.getBasePath(), rollbackRequest.getPartitionPath()), - rollbackRequest.getFileId().get(), HoodieFileFormat.HOODIE_LOG.getFileExtension(), - rollbackRequest.getLatestBaseInstant().get()).collect(Collectors.toMap(HoodieLogFile::getFileStatus, value -> value.getFileStatus().getLen())); + fileId, HoodieFileFormat.HOODIE_LOG.getFileExtension(), latestBaseInstant) + .collect(Collectors.toMap(HoodieLogFile::getFileStatus, value -> value.getFileStatus().getLen())); Writer writer = null; try { writer = HoodieLogFormat.newWriterBuilder() .onParentPath(FSUtils.getPartitionPath(metaClient.getBasePath(), rollbackRequest.getPartitionPath())) - .withFileId(rollbackRequest.getFileId().get()) - .overBaseCommit(rollbackRequest.getLatestBaseInstant().get()).withFs(metaClient.getFs()) + .withFileId(fileId) + .overBaseCommit(latestBaseInstant) + .withFs(metaClient.getFs()) .withFileExtension(HoodieLogFile.DELTA_EXTENSION).build(); // generate metadata diff --git a/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java index 642543d57979f..0a0f59b6cf000 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java @@ -435,21 +435,6 @@ public static Stream getAllLogFiles(FileSystem fs, Path partition } } - /** - * Get all the log files for the passed in FileId in the partition path. - */ - public static Stream getAllLogFiles(FileSystem fs, Path partitionPath, - final String logFileExtension, final String baseCommitTime) throws IOException { - try { - return Arrays - .stream(fs.listStatus(partitionPath, - path -> path.getName().contains(logFileExtension))) - .map(HoodieLogFile::new).filter(s -> s.getBaseCommitTime().equals(baseCommitTime)); - } catch (FileNotFoundException e) { - return Stream.builder().build(); - } - } - /** * Get the latest log version for the fileId in the partition path. */ 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 b3d928749bae1..b9ba11779fa45 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 @@ -44,6 +44,7 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.function.BiFunction; import java.util.stream.Collectors; import static org.apache.hudi.metadata.HoodieTableMetadata.NON_PARTITIONED_NAME; @@ -241,7 +242,7 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad rollbackMetadata.getPartitionMetadata().values().forEach(pm -> { // Has this rollback produced new files? boolean hasRollbackLogFiles = pm.getRollbackLogFiles() != null && !pm.getRollbackLogFiles().isEmpty(); - boolean hasNonZeroRollbackLogFiles = hasRollbackLogFiles ? pm.getRollbackLogFiles().values().stream().mapToLong(Long::longValue).sum() > 0 : false; + boolean hasNonZeroRollbackLogFiles = hasRollbackLogFiles && pm.getRollbackLogFiles().values().stream().mapToLong(Long::longValue).sum() > 0; // If commit being rolled back has not been synced to metadata table yet then there is no need to update metadata boolean shouldSkip = lastSyncTs.isPresent() && HoodieTimeline.compareTimestamps(rollbackMetadata.getCommitsRollback().get(0), HoodieTimeline.GREATER_THAN, lastSyncTs.get()); @@ -264,6 +265,12 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad partitionToDeletedFiles.get(partition).addAll(deletedFiles); } + BiFunction fileMergeFn = (oldSize, newSizeCopy) -> { + // if a file exists in both written log files and rollback log files, we want to pick the one that is higher + // as rollback file could have been updated after written log files are computed. + return oldSize > newSizeCopy ? oldSize : newSizeCopy; + }; + if (hasRollbackLogFiles) { if (!partitionToAppendedFiles.containsKey(partition)) { partitionToAppendedFiles.put(partition, new HashMap<>()); @@ -271,9 +278,7 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad // Extract appended file name from the absolute paths saved in getAppendFiles() pm.getRollbackLogFiles().forEach((path, size) -> { - partitionToAppendedFiles.get(partition).merge(new Path(path).getName(), size, (oldSize, newSizeCopy) -> { - return size + oldSize; - }); + partitionToAppendedFiles.get(partition).merge(new Path(path).getName(), size, fileMergeFn); }); } @@ -284,11 +289,7 @@ private static void processRollbackMetadata(HoodieRollbackMetadata rollbackMetad // Extract appended file name from the absolute paths saved in getWrittenLogFiles() pm.getWrittenLogFiles().forEach((path, size) -> { - partitionToAppendedFiles.get(partition).merge(new Path(path).getName(), size, (oldSize, newSizeCopy) -> { - // if a file exists in both written log files and rollback log files, we want to pick the one that is higher - // as rollback file could have been updated after written log files are computed. - return oldSize > newSizeCopy ? oldSize : newSizeCopy; - }); + partitionToAppendedFiles.get(partition).merge(new Path(path).getName(), size, fileMergeFn); }); } });