You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by si...@apache.org on 2024/05/14 21:18:43 UTC
(hudi) branch branch-0.x updated: [HUDI-1517] create marker file for every log file (#11187)
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/branch-0.x by this push:
new c2c7e0538f8 [HUDI-1517] create marker file for every log file (#11187)
c2c7e0538f8 is described below
commit c2c7e0538f8cf3031781ebdd776d1c03bfec3bb3
Author: Sagar Sumit <sa...@gmail.com>
AuthorDate: Wed May 15 02:48:38 2024 +0530
[HUDI-1517] create marker file for every log file (#11187)
* [HUDI-1517] create marker file for every log file (#4913) (#524)
---------
Co-authored-by: guanziyue <30...@users.noreply.github.com>
Co-authored-by: Lokesh Jain <lj...@apache.org>
---
.../hudi/cli/integ/ITTestMarkersCommand.java | 4 +-
.../apache/hudi/client/BaseHoodieWriteClient.java | 20 +-
.../hudi/client/utils/CommitMetadataUtils.java | 251 ++++++++++++++++++
.../org/apache/hudi/io/HoodieAppendHandle.java | 7 -
.../java/org/apache/hudi/io/HoodieWriteHandle.java | 30 +++
.../metadata/HoodieBackedTableMetadataWriter.java | 5 +-
.../rollback/BaseRollbackActionExecutor.java | 5 +-
.../table/action/rollback/BaseRollbackHelper.java | 200 ++++++++++++--
.../rollback/ListingBasedRollbackStrategy.java | 14 +-
.../rollback/MarkerBasedRollbackStrategy.java | 116 +++++----
.../hudi/table/action/rollback/RollbackUtils.java | 15 ++
.../hudi/table/marker/DirectWriteMarkers.java | 81 ++++--
.../marker/TimelineServerBasedWriteMarkers.java | 22 +-
.../org/apache/hudi/table/marker/WriteMarkers.java | 54 ++--
.../table/upgrade/ZeroToOneUpgradeHandler.java | 5 +-
.../hudi/client/utils/TestCommitMetadataUtils.java | 177 +++++++++++++
.../providers/HoodieMetaClientProvider.java | 14 +
.../java/org/apache/hudi/io/FlinkAppendHandle.java | 27 +-
.../apache/hudi/client/SparkRDDWriteClient.java | 20 +-
.../org/apache/hudi/data/HoodieJavaPairRDD.java | 8 +
.../commit/BaseSparkCommitActionExecutor.java | 5 +
.../BaseSparkDeltaCommitActionExecutor.java | 15 +-
.../functional/TestHoodieBackedMetadata.java | 113 ++++----
.../apache/hudi/data/TestHoodieJavaPairRDD.java | 110 ++++++++
.../hudi/table/TestHoodieMergeOnReadTable.java | 8 +-
.../TestCopyOnWriteRollbackActionExecutor.java | 10 +-
.../TestMergeOnReadRollbackActionExecutor.java | 2 +-
.../table/action/rollback/TestRollbackUtils.java | 38 +++
.../TestHoodieSparkCopyOnWriteTableRollback.java | 65 +++++
...dieSparkMergeOnReadTableInsertUpdateDelete.java | 75 +++++-
.../TestHoodieSparkMergeOnReadTableRollback.java | 206 ++++++++++++---
.../table/functional/TestHoodieSparkRollback.java | 287 +++++++++++++++++++++
.../TestMarkerBasedRollbackStrategy.java | 54 +++-
.../hudi/table/marker/TestWriteMarkersBase.java | 21 ++
.../hudi/table/upgrade/TestUpgradeDowngrade.java | 44 ++++
.../hudi/common/data/HoodieListPairData.java | 27 ++
.../apache/hudi/common/data/HoodiePairData.java | 12 +
.../java/org/apache/hudi/common/fs/FSUtils.java | 58 +++++
.../table/log/HoodieLogFileWriteCallback.java | 42 +++
.../hudi/common/table/log/HoodieLogFormat.java | 15 +-
.../common/table/log/HoodieLogFormatWriter.java | 22 +-
.../hudi/common/table/marker/MarkerOperation.java | 1 +
.../hudi/metadata/HoodieTableMetadataUtil.java | 35 +--
.../common/data/TestHoodieListDataPairData.java | 34 +++
.../org/apache/hudi/common/fs/TestFSUtils.java | 21 ++
.../apache/hudi/common/fs/TestFSUtilsMocked.java | 116 +++++++++
.../hudi/common/testutils/FileCreateUtils.java | 49 +++-
.../hudi/common/testutils/HoodieTestTable.java | 9 +-
.../org/apache/hudi/storage/StorageSchemes.java | 75 +++---
.../sql/hudi/procedure/TestCallProcedure.scala | 6 +-
.../hudi/timeline/service/RequestHandler.java | 7 +
.../timeline/service/handlers/MarkerHandler.java | 10 +
52 files changed, 2318 insertions(+), 349 deletions(-)
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestMarkersCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestMarkersCommand.java
index 25dd3c2152c..df0aa76564b 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestMarkersCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestMarkersCommand.java
@@ -70,8 +70,8 @@ public class ITTestMarkersCommand extends HoodieCLIIntegrationTestBase {
// generate markers
String instantTime1 = "101";
- FileCreateUtils.createMarkerFile(tablePath, "partA", instantTime1, "f0", IOType.APPEND);
- FileCreateUtils.createMarkerFile(tablePath, "partA", instantTime1, "f1", IOType.APPEND);
+ FileCreateUtils.createLogFileMarker(tablePath, "partA", instantTime1, "f0", IOType.APPEND);
+ FileCreateUtils.createLogFileMarker(tablePath, "partA", instantTime1, "f1", IOType.APPEND);
assertEquals(2, FileCreateUtils.getTotalMarkerFileCount(tablePath, "partA", instantTime1, IOType.APPEND));
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
index 1bbf258bae2..52b9fecf658 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
@@ -222,10 +222,11 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient
LOG.info("Committing " + instantTime + " action " + commitActionType);
// Create a Hoodie table which encapsulated the commits and files visible
HoodieTable table = createTable(config, hadoopConf);
- HoodieCommitMetadata metadata = CommitUtils.buildMetadata(stats, partitionToReplaceFileIds,
+ HoodieCommitMetadata originalMetadata = CommitUtils.buildMetadata(stats, partitionToReplaceFileIds,
extraMetadata, operationType, config.getWriteSchema(), commitActionType);
HoodieInstant inflightInstant = new HoodieInstant(State.INFLIGHT, commitActionType, instantTime);
HeartbeatUtils.abortIfHeartbeatExpired(instantTime, table, heartbeatClient, config);
+ HoodieCommitMetadata metadata = reconcileCommitMetadata(table, commitActionType, instantTime, originalMetadata);
this.txnManager.beginTransaction(Option.of(inflightInstant),
lastCompletedTxnAndMetadata.isPresent() ? Option.of(lastCompletedTxnAndMetadata.get().getLeft()) : Option.empty());
try {
@@ -271,25 +272,30 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient
return true;
}
+ protected HoodieCommitMetadata reconcileCommitMetadata(HoodieTable table, String commitActionType, String instantTime, HoodieCommitMetadata originalMetadata) {
+ return originalMetadata;
+ }
+
protected void commit(HoodieTable table, String commitActionType, String instantTime, HoodieCommitMetadata metadata,
List<HoodieWriteStat> stats, HoodieData<WriteStatus> writeStatuses) throws IOException {
LOG.info("Committing " + instantTime + " action " + commitActionType);
HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
+ HoodieCommitMetadata reconciledCommitMetadata = reconcileCommitMetadata(table, commitActionType, instantTime, metadata);
// Finalize write
finalizeWrite(table, instantTime, stats);
// do save internal schema to support Implicitly add columns in write process
- if (!metadata.getExtraMetadata().containsKey(SerDeHelper.LATEST_SCHEMA)
- && metadata.getExtraMetadata().containsKey(SCHEMA_KEY) && table.getConfig().getSchemaEvolutionEnable()) {
- saveInternalSchema(table, instantTime, metadata);
+ if (!reconciledCommitMetadata.getExtraMetadata().containsKey(SerDeHelper.LATEST_SCHEMA)
+ && reconciledCommitMetadata.getExtraMetadata().containsKey(SCHEMA_KEY) && table.getConfig().getSchemaEvolutionEnable()) {
+ saveInternalSchema(table, instantTime, reconciledCommitMetadata);
}
// update Metadata table
- writeTableMetadata(table, instantTime, metadata, writeStatuses);
+ writeTableMetadata(table, instantTime, reconciledCommitMetadata, writeStatuses);
activeTimeline.saveAsComplete(new HoodieInstant(true, commitActionType, instantTime),
- Option.of(getUTF8Bytes(metadata.toJsonString())));
+ Option.of(getUTF8Bytes(reconciledCommitMetadata.toJsonString())));
}
// Save internal schema
- private void saveInternalSchema(HoodieTable table, String instantTime, HoodieCommitMetadata metadata) {
+ protected final void saveInternalSchema(HoodieTable table, String instantTime, HoodieCommitMetadata metadata) {
TableSchemaResolver schemaUtil = new TableSchemaResolver(table.getMetaClient());
String historySchemaStr = schemaUtil.getTableHistorySchemaStrFromCommitMetadata().orElse("");
FileBasedInternalSchemaStorageManager schemasManager = new FileBasedInternalSchemaStorageManager(table.getMetaClient());
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/CommitMetadataUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/CommitMetadataUtils.java
new file mode 100644
index 00000000000..8c815e20344
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/CommitMetadataUtils.java
@@ -0,0 +1,251 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.client.utils;
+
+import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.data.HoodiePairData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.function.SerializableBiFunction;
+import org.apache.hudi.common.function.SerializableFunction;
+import org.apache.hudi.common.function.SerializablePairFunction;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieDeltaWriteStat;
+import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.marker.WriteMarkers;
+import org.apache.hudi.table.marker.WriteMarkersFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class CommitMetadataUtils {
+
+ /* In spark mor table, task retries may generate log files which are not included in write status.
+ * We need to add these to CommitMetadata so that it will be synced to MDT.
+ */
+ public static HoodieCommitMetadata reconcileMetadataForMissingFiles(HoodieTable table, String commitActionType, String instantTime,
+ HoodieCommitMetadata commitMetadata, HoodieWriteConfig config,
+ HoodieEngineContext context, Configuration hadoopConf,
+ String classNameForContext) throws IOException {
+ if (!table.getMetaClient().getTableType().equals(HoodieTableType.MERGE_ON_READ)
+ || !commitActionType.equals(HoodieActiveTimeline.DELTA_COMMIT_ACTION)) {
+ return commitMetadata;
+ }
+
+ WriteMarkers markers = WriteMarkersFactory.get(config.getMarkersType(), table, instantTime);
+ // if there is log files in this delta commit, we search any invalid log files generated by failed spark task
+ boolean hasLogFileInDeltaCommit = commitMetadata.getPartitionToWriteStats()
+ .values().stream().flatMap(List::stream)
+ .anyMatch(writeStat -> FSUtils.isLogFile(new Path(config.getBasePath(), writeStat.getPath()).getName()));
+ if (hasLogFileInDeltaCommit) { // skip for COW table
+ // get all log files generated by makers
+ Set<String> allLogFilesMarkerPath = new HashSet<>(markers.getAppendedLogPaths(context, config.getFinalizeWriteParallelism()));
+ Set<String> logFilesMarkerPath = new HashSet<>();
+ allLogFilesMarkerPath.stream().filter(logFilePath -> !logFilePath.endsWith("cdc")).forEach(logFilesMarkerPath::add);
+
+ // remove valid log files
+ // TODO: refactor based on HoodieData
+ for (Map.Entry<String, List<HoodieWriteStat>> partitionAndWriteStats : commitMetadata.getPartitionToWriteStats().entrySet()) {
+ for (HoodieWriteStat hoodieWriteStat : partitionAndWriteStats.getValue()) {
+ logFilesMarkerPath.remove(hoodieWriteStat.getPath());
+ }
+ }
+
+ // remaining are log files generated by retried spark task, let's generate write stat for them
+ if (!logFilesMarkerPath.isEmpty()) {
+ SerializableConfiguration serializableConfiguration = new SerializableConfiguration(hadoopConf);
+ context.setJobStatus(classNameForContext, "Preparing data for missing files to assist with generating write stats");
+ // populate partition -> map (fileId -> HoodieWriteStat) // we just need one write stat per fileID to fetch some info about
+ // the file slice of interest to populate WriteStat.
+ HoodiePairData<String, Map<String, HoodieWriteStat>> partitionToWriteStatHoodieData = getPartitionToFileIdToFilesMap(commitMetadata, context);
+
+ String basePathStr = config.getBasePath();
+ // populate partition -> map (fileId -> List <missing log file names>)
+ HoodiePairData<String, Map<String, List<String>>> partitionToMissingLogFilesHoodieData =
+ getPartitionToFileIdToMissingLogFileMap(basePathStr, logFilesMarkerPath, context, config.getFileListingParallelism());
+
+ context.setJobStatus(classNameForContext, "Generating writeStat for missing log files");
+
+ // lets join both to generate write stats for missing log files
+ List<Pair<String, List<HoodieWriteStat>>> additionalLogFileWriteStat = getWriteStatsForMissingLogFiles(partitionToWriteStatHoodieData,
+ partitionToMissingLogFilesHoodieData, serializableConfiguration, basePathStr);
+
+ for (Pair<String, List<HoodieWriteStat>> partitionDeltaStats : additionalLogFileWriteStat) {
+ String partitionPath = partitionDeltaStats.getKey();
+ partitionDeltaStats.getValue().forEach(ws -> commitMetadata.addWriteStat(partitionPath, ws));
+ }
+ }
+ }
+ return commitMetadata;
+ }
+
+ /**
+ * Get partition path to fileId to write stat map.
+ */
+ private static HoodiePairData<String, Map<String, HoodieWriteStat>> getPartitionToFileIdToFilesMap(HoodieCommitMetadata commitMetadata, HoodieEngineContext context) {
+ List<Map.Entry<String, List<HoodieWriteStat>>> partitionToWriteStats = new ArrayList<>(commitMetadata.getPartitionToWriteStats().entrySet());
+
+ return context.parallelize(partitionToWriteStats)
+ .mapToPair((SerializablePairFunction<Map.Entry<String, List<HoodieWriteStat>>, String, Map<String, HoodieWriteStat>>) t -> {
+ Map<String, HoodieWriteStat> fileIdToWriteStat = new HashMap<>();
+ t.getValue().forEach(writeStat -> {
+ if (!fileIdToWriteStat.containsKey(writeStat.getFileId())) {
+ fileIdToWriteStat.put(writeStat.getFileId(), writeStat);
+ }
+ });
+ return Pair.of(t.getKey(), fileIdToWriteStat);
+ });
+ }
+
+ /**
+ * Get partition path to fileId to missing log file map.
+ *
+ * @param basePathStr base path
+ * @param logFilesMarkerPath set of log file marker paths
+ * @param context HoodieEngineContext
+ * @param parallelism parallelism
+ * @return HoodiePairData of partition path to fileId to missing log file map.
+ */
+ private static HoodiePairData<String, Map<String, List<String>>> getPartitionToFileIdToMissingLogFileMap(String basePathStr, Set<String> logFilesMarkerPath, HoodieEngineContext context,
+ int parallelism) {
+ List<String> logFilePaths = new ArrayList<>(logFilesMarkerPath);
+ HoodiePairData<String, List<String>> partitionPathLogFilePair = context.parallelize(logFilePaths).mapToPair(logFilePath -> {
+ Path logFileFullPath = new Path(basePathStr, logFilePath);
+ String partitionPath = FSUtils.getRelativePartitionPath(new Path(basePathStr), logFileFullPath.getParent());
+ return Pair.of(partitionPath, Collections.singletonList(logFileFullPath.getName()));
+ });
+ HoodiePairData<String, Map<String, List<String>>> partitionPathToFileIdAndLogFileList = partitionPathLogFilePair
+ // reduce by partition paths
+ .reduceByKey((SerializableBiFunction<List<String>, List<String>, List<String>>) (strings, strings2) -> {
+ List<String> logFilePaths1 = new ArrayList<>(strings);
+ logFilePaths1.addAll(strings2);
+ return logFilePaths1;
+ }, parallelism).mapToPair((SerializablePairFunction<Pair<String, List<String>>, String, Map<String, List<String>>>) t -> {
+ // for each hudi partition, collect list of missing log files, fetch file size using file system calls, and populate fileId -> List<FileStatus> map
+
+ String partitionPath = t.getKey();
+ Path fullPartitionPath = StringUtils.isNullOrEmpty(partitionPath) ? new Path(basePathStr) : new Path(basePathStr, partitionPath);
+ // fetch file sizes from FileSystem
+ List<String> missingLogFiles = t.getValue();
+ Map<String, List<String>> fileIdtologFiles = new HashMap<>();
+ missingLogFiles.forEach(logFile -> {
+ String fileId = FSUtils.getFileIdFromLogPath(new Path(fullPartitionPath, logFile));
+ if (!fileIdtologFiles.containsKey(fileId)) {
+ fileIdtologFiles.put(fileId, new ArrayList<>());
+ }
+ fileIdtologFiles.get(fileId).add(logFile);
+ });
+ return Pair.of(partitionPath, fileIdtologFiles);
+ });
+ return partitionPathToFileIdAndLogFileList;
+ }
+
+ /**
+ * Generate write stats for missing log files. Performs an inner join on partition between existing
+ * partitionToWriteStatHoodieData and partitionToMissingLogFilesHoodieData.
+ * For missing log files, it does one file system call to fetch file size (FSUtils#getFileStatusesUnderPartition).
+ */
+ private static List<Pair<String, List<HoodieWriteStat>>> getWriteStatsForMissingLogFiles(HoodiePairData<String, Map<String, HoodieWriteStat>> partitionToWriteStatHoodieData,
+ HoodiePairData<String, Map<String, List<String>>> partitionToMissingLogFilesHoodieData,
+ SerializableConfiguration serializableConfiguration,
+ String basePathStr) {
+ // lets join both to generate write stats for missing log files
+ return partitionToWriteStatHoodieData
+ .join(partitionToMissingLogFilesHoodieData)
+ .map((SerializableFunction<Pair<String, Pair<Map<String, HoodieWriteStat>, Map<String, List<String>>>>, Pair<String, List<HoodieWriteStat>>>) v1 -> {
+ final Path basePathLocal = new Path(basePathStr);
+ String partitionPath = v1.getKey();
+ Map<String, HoodieWriteStat> fileIdToOriginalWriteStat = v1.getValue().getKey();
+ Map<String, List<String>> missingFileIdToLogFileNames = v1.getValue().getValue();
+ List<String> missingLogFileNames = missingFileIdToLogFileNames.values().stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList());
+
+ // fetch file sizes from FileSystem
+ Path fullPartitionPath = StringUtils.isNullOrEmpty(partitionPath) ? new Path(basePathStr) : new Path(basePathStr, partitionPath);
+ FileSystem fileSystem = fullPartitionPath.getFileSystem(serializableConfiguration.get());
+ List<Option<FileStatus>> fileStatuesOpt = FSUtils.getFileStatusesUnderPartition(fileSystem, fullPartitionPath, new HashSet<>(missingLogFileNames), true);
+ List<FileStatus> fileStatuses = fileStatuesOpt.stream().filter(fileStatusOpt -> fileStatusOpt.isPresent()).map(fileStatusOption -> fileStatusOption.get()).collect(Collectors.toList());
+
+ // populate fileId -> List<FileStatus>
+ Map<String, List<FileStatus>> missingFileIdToLogFilesList = new HashMap<>();
+ fileStatuses.forEach(fileStatus -> {
+ String fileId = FSUtils.getFileIdFromLogPath(fileStatus.getPath());
+ missingFileIdToLogFilesList.putIfAbsent(fileId, new ArrayList<>());
+ missingFileIdToLogFilesList.get(fileId).add(fileStatus);
+ });
+
+ List<HoodieWriteStat> missingWriteStats = new ArrayList();
+ missingFileIdToLogFilesList.forEach((k, logFileStatuses) -> {
+ String fileId = k;
+ HoodieDeltaWriteStat originalWriteStat =
+ (HoodieDeltaWriteStat) fileIdToOriginalWriteStat.get(fileId); // are there chances that there won't be any write stat in original list?
+ logFileStatuses.forEach(fileStatus -> {
+ // for every missing file, add a new HoodieDeltaWriteStat
+ HoodieDeltaWriteStat writeStat = getHoodieDeltaWriteStatFromPreviousStat(fileStatus, basePathLocal,
+ partitionPath, fileId, originalWriteStat);
+ missingWriteStats.add(writeStat);
+ });
+ });
+ return Pair.of(partitionPath, missingWriteStats);
+ }).collectAsList();
+ }
+
+ private static HoodieDeltaWriteStat getHoodieDeltaWriteStatFromPreviousStat(FileStatus fileStatus,
+ Path basePathLocal,
+ String partitionPath,
+ String fileId,
+ HoodieDeltaWriteStat originalWriteStat) {
+ HoodieDeltaWriteStat writeStat = new HoodieDeltaWriteStat();
+ HoodieLogFile logFile = new HoodieLogFile(fileStatus);
+ writeStat.setPath(basePathLocal, logFile.getPath());
+ writeStat.setPartitionPath(partitionPath);
+ writeStat.setFileId(fileId);
+ writeStat.setTotalWriteBytes(logFile.getFileSize());
+ writeStat.setFileSizeInBytes(logFile.getFileSize());
+ writeStat.setLogVersion(logFile.getLogVersion());
+ List<String> logFiles = new ArrayList<>(originalWriteStat.getLogFiles());
+ logFiles.add(logFile.getFileName());
+ writeStat.setLogFiles(logFiles);
+ writeStat.setBaseFile(originalWriteStat.getBaseFile());
+ writeStat.setPrevCommit(logFile.getBaseCommitTime());
+ return writeStat;
+ }
+}
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 5d9c5ac5496..aab6ecbe735 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
@@ -210,13 +210,6 @@ public class HoodieAppendHandle<T, I, K, O> extends HoodieWriteHandle<T, I, K, O
new Path(config.getBasePath()), FSUtils.getPartitionPath(config.getBasePath(), partitionPath),
hoodieTable.getPartitionMetafileFormat());
partitionMetadata.trySave(getPartitionId());
-
- // 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.
- // https://issues.apache.org/jira/browse/HUDI-1517
- createMarkerFile(partitionPath, FSUtils.makeBaseFileName(baseInstantTime, writeToken, fileId, hoodieTable.getBaseFileExtension()));
-
this.writer = createLogWriter(fileSlice, baseInstantTime);
} catch (Exception e) {
LOG.error("Error in update task at commit " + instantTime, e);
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java
index 81480767599..0aecb2c087c 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java
@@ -31,6 +31,7 @@ import org.apache.hudi.common.model.HoodieRecordLocation;
import org.apache.hudi.common.model.HoodieRecordMerger;
import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.log.HoodieLogFileWriteCallback;
import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.common.util.Option;
@@ -39,6 +40,7 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.marker.WriteMarkers;
import org.apache.hudi.table.marker.WriteMarkersFactory;
import org.apache.avro.Schema;
@@ -255,9 +257,14 @@ public abstract class HoodieWriteHandle<T, I, K, O> extends HoodieIOHandle<T, I,
.withRolloverLogWriteToken(writeToken)
.withLogWriteToken(latestLogFile.map(HoodieLogFile::getLogWriteToken).orElse(writeToken))
.withSuffix(suffix)
+ .withLogWriteCallback(getLogWriteCallback())
.withFileExtension(HoodieLogFile.DELTA_EXTENSION).build();
}
+ protected HoodieLogFileWriteCallback getLogWriteCallback() {
+ return new AppendLogWriteCallback();
+ }
+
protected HoodieLogFormat.Writer createLogWriter(String baseCommitTime, String fileSuffix) {
try {
return createLogWriter(Option.empty(),baseCommitTime, fileSuffix);
@@ -276,4 +283,27 @@ public abstract class HoodieWriteHandle<T, I, K, O> extends HoodieIOHandle<T, I,
return Option.empty();
}
}
+
+ /**
+ * Call back to be invoked during log file creation and appends. Applicable only for AppendHandle among all write handles.
+ */
+ protected class AppendLogWriteCallback implements HoodieLogFileWriteCallback {
+
+ @Override
+ public boolean preLogFileOpen(HoodieLogFile logFileToAppend) {
+ return createAppendMarker(logFileToAppend);
+ }
+
+ @Override
+ public boolean preLogFileCreate(HoodieLogFile logFileToCreate) {
+ // TODO: HUDI-1517 may distinguish log file created from log file being appended in the future @guanziyue
+ return createAppendMarker(logFileToCreate);
+ }
+
+ private boolean createAppendMarker(HoodieLogFile logFileToAppend) {
+ WriteMarkers writeMarkers = WriteMarkersFactory.get(config.getMarkersType(), hoodieTable, instantTime);
+ return writeMarkers.createIfNotExists(partitionPath, logFileToAppend.getFileName(), IOType.APPEND,
+ config, fileId, hoodieTable.getMetaClient().getActiveTimeline()).isPresent();
+ }
+ }
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
index e508e2d2b7e..e8dd6021498 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieBackedTableMetadataWriter.java
@@ -992,14 +992,13 @@ public abstract class HoodieBackedTableMetadataWriter<I> implements HoodieTableM
validateRollback(commitToRollbackInstantTime, compactionInstant, deltacommitsSinceCompaction);
- // lets apply a delta commit with DT's rb instant(with special suffix) containing following records:
+ // lets apply a delta commit with DT's rb instant containing following records:
// a. any log files as part of RB commit metadata that was added
// b. log files added by the commit in DT being rolled back. By rolled back, we mean, a rollback block will be added and does not mean it will be deleted.
// both above list should only be added to FILES partition.
-
- String rollbackInstantTime = createRollbackTimestamp(instantTime);
processAndCommit(instantTime, () -> HoodieTableMetadataUtil.convertMetadataToRecords(engineContext, dataMetaClient, rollbackMetadata, instantTime));
+ String rollbackInstantTime = createRollbackTimestamp(instantTime);
if (deltacommitsSinceCompaction.containsInstant(deltaCommitInstant)) {
LOG.info("Rolling back MDT deltacommit " + commitToRollbackInstantTime);
if (!getWriteClient().rollback(commitToRollbackInstantTime, rollbackInstantTime)) {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java
index 662bfe36299..f2a40512b88 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java
@@ -119,6 +119,9 @@ public abstract class BaseRollbackActionExecutor<T, I, K, O> extends BaseActionE
// Finally, remove the markers post rollback.
WriteMarkersFactory.get(config.getMarkersType(), table, instantToRollback.getTimestamp())
.quietDeleteMarkerDir(context, config.getMarkersDeleteParallelism());
+ // For MOR table rollbacks, rollback command blocks might generate markers under rollback instant. So, lets clean up the markers if any.
+ WriteMarkersFactory.get(config.getMarkersType(), table, rollbackInstant.getTimestamp())
+ .quietDeleteMarkerDir(context, config.getMarkersDeleteParallelism());
return rollbackMetadata;
}
@@ -239,7 +242,7 @@ public abstract class BaseRollbackActionExecutor<T, I, K, O> extends BaseActionE
* @return list of {@link HoodieRollbackStat}s.
*/
protected List<HoodieRollbackStat> executeRollback(HoodieInstant instantToRollback, HoodieRollbackPlan rollbackPlan) {
- return new BaseRollbackHelper(table.getMetaClient(), config).performRollback(context, instantToRollback, rollbackPlan.getRollbackRequests());
+ return new BaseRollbackHelper(table, config).performRollback(context, instantTime, instantToRollback, rollbackPlan.getRollbackRequests());
}
protected void finishRollback(HoodieInstant inflightInstant, HoodieRollbackMetadata rollbackMetadata) throws HoodieIOException {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackHelper.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackHelper.java
index 94473e98d79..d2014bbb808 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackHelper.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackHelper.java
@@ -20,21 +20,32 @@ package org.apache.hudi.table.action.rollback;
import org.apache.hudi.avro.model.HoodieRollbackRequest;
import org.apache.hudi.common.HoodieRollbackStat;
+import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.data.HoodiePairData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.function.SerializableFunction;
+import org.apache.hudi.common.function.SerializablePairFunction;
import org.apache.hudi.common.model.HoodieLogFile;
+import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.log.HoodieLogFileWriteCallback;
import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.block.HoodieCommandBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieRollbackException;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.marker.WriteMarkers;
+import org.apache.hudi.table.marker.WriteMarkersFactory;
import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,9 +56,11 @@ import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -60,18 +73,20 @@ public class BaseRollbackHelper implements Serializable {
private static final Logger LOG = LoggerFactory.getLogger(BaseRollbackHelper.class);
protected static final String EMPTY_STRING = "";
+ protected final HoodieTable table;
protected final HoodieTableMetaClient metaClient;
protected final HoodieWriteConfig config;
- public BaseRollbackHelper(HoodieTableMetaClient metaClient, HoodieWriteConfig config) {
- this.metaClient = metaClient;
+ public BaseRollbackHelper(HoodieTable table, HoodieWriteConfig config) {
+ this.table = table;
+ this.metaClient = table.getMetaClient();
this.config = config;
}
/**
* Performs all rollback actions that we have collected in parallel.
*/
- public List<HoodieRollbackStat> performRollback(HoodieEngineContext context, HoodieInstant instantToRollback,
+ public List<HoodieRollbackStat> performRollback(HoodieEngineContext context, String instantTime, HoodieInstant instantToRollback,
List<HoodieRollbackRequest> rollbackRequests) {
int parallelism = Math.max(Math.min(rollbackRequests.size(), config.getRollbackParallelism()), 1);
context.setJobStatus(this.getClass().getSimpleName(), "Perform rollback actions: " + config.getTableName());
@@ -80,14 +95,28 @@ public class BaseRollbackHelper implements Serializable {
// stack trace: https://gist.github.com/nsivabalan/b6359e7d5038484f8043506c8bc9e1c8
// related stack overflow post: https://issues.apache.org/jira/browse/SPARK-3601. Avro deserializes list as GenericData.Array.
List<SerializableHoodieRollbackRequest> serializableRequests = rollbackRequests.stream().map(SerializableHoodieRollbackRequest::new).collect(Collectors.toList());
- return context.reduceByKey(maybeDeleteAndCollectStats(context, instantToRollback, serializableRequests, true, parallelism),
- RollbackUtils::mergeRollbackStat, parallelism);
+ WriteMarkers markers = WriteMarkersFactory.get(config.getMarkersType(), table, instantTime);
+
+ // Considering rollback may failed before, which generated some additional log files. We need to add these log files back.
+ // rollback markers are added under rollback instant itself.
+ Set<String> logPaths = new HashSet<>();
+ try {
+ logPaths = markers.getAppendedLogPaths(context, config.getFinalizeWriteParallelism());
+ } catch (FileNotFoundException fnf) {
+ LOG.warn("Rollback never failed and hence no marker dir was found. Safely moving on");
+ } catch (IOException e) {
+ throw new HoodieRollbackException("Failed to list log file markers for previous attempt of rollback ", e);
+ }
+
+ List<Pair<String, HoodieRollbackStat>> getRollbackStats = maybeDeleteAndCollectStats(context, instantTime, instantToRollback, serializableRequests, true, parallelism);
+ List<HoodieRollbackStat> mergedRollbackStatByPartitionPath = context.reduceByKey(getRollbackStats, RollbackUtils::mergeRollbackStat, parallelism);
+ return addLogFilesFromPreviousFailedRollbacksToStat(context, mergedRollbackStatByPartitionPath, logPaths);
}
/**
* Collect all file info that needs to be rolled back.
*/
- public List<HoodieRollbackStat> collectRollbackStats(HoodieEngineContext context, HoodieInstant instantToRollback,
+ public List<HoodieRollbackStat> collectRollbackStats(HoodieEngineContext context, String instantTime, HoodieInstant instantToRollback,
List<HoodieRollbackRequest> rollbackRequests) {
int parallelism = Math.max(Math.min(rollbackRequests.size(), config.getRollbackParallelism()), 1);
context.setJobStatus(this.getClass().getSimpleName(), "Collect rollback stats for upgrade/downgrade: " + config.getTableName());
@@ -96,7 +125,7 @@ public class BaseRollbackHelper implements Serializable {
// stack trace: https://gist.github.com/nsivabalan/b6359e7d5038484f8043506c8bc9e1c8
// related stack overflow post: https://issues.apache.org/jira/browse/SPARK-3601. Avro deserializes list as GenericData.Array.
List<SerializableHoodieRollbackRequest> serializableRequests = rollbackRequests.stream().map(SerializableHoodieRollbackRequest::new).collect(Collectors.toList());
- return context.reduceByKey(maybeDeleteAndCollectStats(context, instantToRollback, serializableRequests, false, parallelism),
+ return context.reduceByKey(maybeDeleteAndCollectStats(context, instantTime, instantToRollback, serializableRequests, false, parallelism),
RollbackUtils::mergeRollbackStat, parallelism);
}
@@ -110,6 +139,7 @@ public class BaseRollbackHelper implements Serializable {
* @return stats collected with or w/o actual deletions.
*/
List<Pair<String, HoodieRollbackStat>> maybeDeleteAndCollectStats(HoodieEngineContext context,
+ String instantTime,
HoodieInstant instantToRollback,
List<SerializableHoodieRollbackRequest> rollbackRequests,
boolean doDelete, int numPartitions) {
@@ -124,14 +154,19 @@ public class BaseRollbackHelper implements Serializable {
HoodieLogFormat.Writer writer = null;
final Path filePath;
try {
+ String partitionPath = rollbackRequest.getPartitionPath();
String fileId = rollbackRequest.getFileId();
String latestBaseInstant = rollbackRequest.getLatestBaseInstant();
+ // Let's emit markers for rollback as well. markers are emitted under rollback instant time.
+ WriteMarkers writeMarkers = WriteMarkersFactory.get(config.getMarkersType(), table, instantTime);
+
writer = HoodieLogFormat.newWriterBuilder()
- .onParentPath(FSUtils.getPartitionPath(metaClient.getBasePath(), rollbackRequest.getPartitionPath()))
+ .onParentPath(FSUtils.getPartitionPath(metaClient.getBasePathV2().toString(), partitionPath))
.withFileId(fileId)
.overBaseCommit(latestBaseInstant)
.withFs(metaClient.getFs())
+ .withLogWriteCallback(getRollbackLogMarkerCallback(writeMarkers, partitionPath, fileId))
.withFileExtension(HoodieLogFile.DELTA_EXTENSION).build();
// generate metadata
@@ -163,30 +198,159 @@ public class BaseRollbackHelper implements Serializable {
1L
);
+ // With listing based rollback, sometimes we only get the fileID of interest(so that we can add rollback command block) w/o the actual file name.
+ // So, we want to ignore such invalid files from this list before we add it to the rollback stats.
+ String partitionFullPath = FSUtils.getPartitionPath(metaClient.getBasePathV2().toString(), rollbackRequest.getPartitionPath()).toString();
+ Map<String, Long> validLogBlocksToDelete = new HashMap<>();
+ rollbackRequest.getLogBlocksToBeDeleted().entrySet().stream().forEach((kv) -> {
+ String logFileFullPath = kv.getKey();
+ String logFileName = logFileFullPath.replace(partitionFullPath, "");
+ if (!StringUtils.isNullOrEmpty(logFileName)) {
+ validLogBlocksToDelete.put(kv.getKey(), kv.getValue());
+ }
+ });
+
return Collections.singletonList(
- Pair.of(rollbackRequest.getPartitionPath(),
- HoodieRollbackStat.newBuilder()
- .withPartitionPath(rollbackRequest.getPartitionPath())
- .withRollbackBlockAppendResults(filesToNumBlocksRollback)
- .build()))
+ Pair.of(rollbackRequest.getPartitionPath(),
+ HoodieRollbackStat.newBuilder()
+ .withPartitionPath(rollbackRequest.getPartitionPath())
+ .withRollbackBlockAppendResults(filesToNumBlocksRollback)
+ .withLogFilesFromFailedCommit(validLogBlocksToDelete)
+ .build()))
.stream();
} else {
return Collections.singletonList(
- Pair.of(rollbackRequest.getPartitionPath(),
- HoodieRollbackStat.newBuilder()
- .withPartitionPath(rollbackRequest.getPartitionPath())
- .build()))
+ Pair.of(rollbackRequest.getPartitionPath(),
+ HoodieRollbackStat.newBuilder()
+ .withPartitionPath(rollbackRequest.getPartitionPath())
+ .build()))
.stream();
}
}, numPartitions);
}
+ private HoodieLogFileWriteCallback getRollbackLogMarkerCallback(final WriteMarkers writeMarkers, String partitionPath, String fileId) {
+ return new HoodieLogFileWriteCallback() {
+ @Override
+ public boolean preLogFileOpen(HoodieLogFile logFileToAppend) {
+ // there may be existed marker file if fs support append. So always return true;
+ createAppendMarker(logFileToAppend);
+ return true;
+ }
+
+ @Override
+ public boolean preLogFileCreate(HoodieLogFile logFileToCreate) {
+ return createAppendMarker(logFileToCreate);
+ }
+
+ private boolean createAppendMarker(HoodieLogFile logFileToAppend) {
+ return writeMarkers.createIfNotExists(partitionPath, logFileToAppend.getFileName(), IOType.APPEND,
+ config, fileId, metaClient.getActiveTimeline()).isPresent();
+ }
+ };
+ }
+
+ /**
+ * If there are log files created by previous rollback attempts, we want to add them to rollback stats so that MDT is able to track them.
+ * @param context HoodieEngineContext
+ * @param originalRollbackStats original rollback stats
+ * @param logPaths log paths due to failed rollback attempts
+ * @return
+ */
+ private List<HoodieRollbackStat> addLogFilesFromPreviousFailedRollbacksToStat(HoodieEngineContext context,
+ List<HoodieRollbackStat> originalRollbackStats,
+ Set<String> logPaths) {
+ if (logPaths.isEmpty()) {
+ // if rollback is not failed and re-attempted, we should not find any additional log files here.
+ return originalRollbackStats;
+ }
+
+ final String basePathStr = metaClient.getBasePathV2().toString();
+ List<String> logFiles = new ArrayList<>(logPaths);
+ // populate partitionPath -> List<log file name>
+ HoodiePairData<String, List<String>> partitionPathToLogFilesHoodieData = populatePartitionToLogFilesHoodieData(context, basePathStr, logFiles);
+
+ // populate partitionPath -> HoodieRollbackStat
+ HoodiePairData<String, HoodieRollbackStat> partitionPathToRollbackStatsHoodieData =
+ context.parallelize(originalRollbackStats)
+ .mapToPair((SerializablePairFunction<HoodieRollbackStat, String, HoodieRollbackStat>) t -> Pair.of(t.getPartitionPath(), t));
+
+ SerializableConfiguration serializableConfiguration = new SerializableConfiguration(context.getHadoopConf());
+
+ // lets do left outer join and append missing log files to HoodieRollbackStat for each partition path.
+ List<HoodieRollbackStat> finalRollbackStats = addMissingLogFilesAndGetRollbackStats(partitionPathToRollbackStatsHoodieData,
+ partitionPathToLogFilesHoodieData, basePathStr, serializableConfiguration);
+ return finalRollbackStats;
+ }
+
+ private HoodiePairData<String, List<String>> populatePartitionToLogFilesHoodieData(HoodieEngineContext context, String basePathStr, List<String> logFiles) {
+ return context.parallelize(logFiles)
+ // lets map each log file to partition path and log file name
+ .mapToPair((SerializablePairFunction<String, String, String>) t -> {
+ Path logFilePath = new Path(basePathStr, t);
+ String partitionPath = FSUtils.getRelativePartitionPath(new Path(basePathStr), logFilePath.getParent());
+ return Pair.of(partitionPath, logFilePath.getName());
+ })
+ // lets group by partition path and collect it as log file list per partition path
+ .groupByKey().mapToPair((SerializablePairFunction<Pair<String, Iterable<String>>, String, List<String>>) t -> {
+ List<String> allFiles = new ArrayList<>();
+ t.getRight().forEach(entry -> allFiles.add(entry));
+ return Pair.of(t.getKey(), allFiles);
+ });
+ }
+
+ /**
+ * Add missing log files to HoodieRollbackStat for each partition path. Performs a left outer join on the partition
+ * key between partitionPathToRollbackStatsHoodieData and partitionPathToLogFilesHoodieData to add the rollback
+ * stats for missing log files.
+ *
+ * @param partitionPathToRollbackStatsHoodieData HoodieRollbackStat by partition path
+ * @param partitionPathToLogFilesHoodieData list of missing log files by partition path
+ * @param basePathStr base path
+ * @param serializableConfiguration hadoop configuration
+ * @return
+ */
+ private List<HoodieRollbackStat> addMissingLogFilesAndGetRollbackStats(HoodiePairData<String, HoodieRollbackStat> partitionPathToRollbackStatsHoodieData,
+ HoodiePairData<String, List<String>> partitionPathToLogFilesHoodieData,
+ String basePathStr, SerializableConfiguration serializableConfiguration) {
+ return partitionPathToRollbackStatsHoodieData
+ .leftOuterJoin(partitionPathToLogFilesHoodieData)
+ .map((SerializableFunction<Pair<String, Pair<HoodieRollbackStat, Option<List<String>>>>, HoodieRollbackStat>) v1 -> {
+ if (v1.getValue().getValue().isPresent()) {
+
+ String partition = v1.getKey();
+ HoodieRollbackStat rollbackStat = v1.getValue().getKey();
+ List<String> missingLogFiles = v1.getValue().getRight().get();
+
+ // fetch file sizes.
+ Path fullPartitionPath = StringUtils.isNullOrEmpty(partition) ? new Path(basePathStr) : new Path(basePathStr, partition);
+ FileSystem fs = fullPartitionPath.getFileSystem(serializableConfiguration.get());
+ List<Option<FileStatus>> fileStatusesOpt = FSUtils.getFileStatusesUnderPartition(fs,
+ fullPartitionPath, new HashSet<>(missingLogFiles), true);
+ List<FileStatus> fileStatuses = fileStatusesOpt.stream().filter(fileStatusOption -> fileStatusOption.isPresent())
+ .map(fileStatusOption -> fileStatusOption.get()).collect(Collectors.toList());
+
+ HashMap<FileStatus, Long> commandBlocksCount = new HashMap<>(rollbackStat.getCommandBlocksCount());
+ fileStatuses.forEach(fileStatus -> commandBlocksCount.put(fileStatus, fileStatus.getLen()));
+
+ return new HoodieRollbackStat(
+ rollbackStat.getPartitionPath(),
+ rollbackStat.getSuccessDeleteFiles(),
+ rollbackStat.getFailedDeleteFiles(),
+ commandBlocksCount,
+ rollbackStat.getLogFilesFromFailedCommit());
+ } else {
+ return v1.getValue().getKey();
+ }
+ }).collectAsList();
+ }
+
/**
* Common method used for cleaning out files during rollback.
*/
protected List<HoodieRollbackStat> deleteFiles(HoodieTableMetaClient metaClient, List<String> filesToBeDeleted, boolean doDelete) throws IOException {
return filesToBeDeleted.stream().map(fileToDelete -> {
- String basePath = metaClient.getBasePath();
+ String basePath = metaClient.getBasePathV2().toString();
try {
Path fullDeletePath = new Path(fileToDelete);
String partitionPath = FSUtils.getRelativePartitionPath(new Path(basePath), fullDeletePath.getParent());
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackStrategy.java
index a622c5ae433..bb7a4235bbb 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackStrategy.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackStrategy.java
@@ -31,6 +31,7 @@ import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieRollbackException;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
@@ -169,7 +170,7 @@ public class ListingBasedRollbackStrategy implements BaseRollbackPlanActionExecu
// (B.4) Rollback triggered for recurring commits - Same as (B.2) plus we need to delete the log files
// as well if the base file gets deleted.
HoodieCommitMetadata commitMetadata = HoodieCommitMetadata.fromBytes(
- table.getMetaClient().getCommitTimeline().getInstantDetails(instantToRollback).get(),
+ table.getMetaClient().getCommitsTimeline().getInstantDetails(instantToRollback).get(),
HoodieCommitMetadata.class);
// In case all data was inserts and the commit failed, delete the file belonging to that commit
@@ -350,20 +351,17 @@ public class ListingBasedRollbackStrategy implements BaseRollbackPlanActionExecu
})
.collect(Collectors.toList());
- for (HoodieWriteStat writeStat : hoodieWriteStats) {
+ for (HoodieWriteStat writeStat : hoodieWriteStats.stream().filter(
+ hoodieWriteStat -> !StringUtils.isNullOrEmpty(hoodieWriteStat.getFileId())).collect(Collectors.toList())) {
FileSlice latestFileSlice = latestFileSlices.get(writeStat.getFileId());
String fileId = writeStat.getFileId();
String latestBaseInstant = latestFileSlice.getBaseInstantTime();
-
Path fullLogFilePath = FSUtils.getPartitionPath(table.getConfig().getBasePath(), writeStat.getPath());
-
- Map<String, Long> logFilesWithBlocksToRollback =
- Collections.singletonMap(fullLogFilePath.toString(), writeStat.getTotalWriteBytes());
-
+ Map<String, Long> logFilesWithBlocksToRollback = Collections.singletonMap(
+ fullLogFilePath.toString(), writeStat.getTotalWriteBytes() > 0 ? writeStat.getTotalWriteBytes() : 1L);
hoodieRollbackRequests.add(new HoodieRollbackRequest(partitionPath, fileId, latestBaseInstant,
Collections.emptyList(), logFilesWithBlocksToRollback));
}
-
return hoodieRollbackRequests;
}
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/MarkerBasedRollbackStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/MarkerBasedRollbackStrategy.java
index 791191c0ef3..431a2f0554a 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/MarkerBasedRollbackStrategy.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/MarkerBasedRollbackStrategy.java
@@ -21,12 +21,14 @@ package org.apache.hudi.table.action.rollback;
import org.apache.hudi.avro.model.HoodieRollbackRequest;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieBaseFile;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieRollbackException;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.marker.MarkerBasedRollbackUtils;
@@ -41,30 +43,27 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
-import static org.apache.hudi.table.action.rollback.BaseRollbackHelper.EMPTY_STRING;
+import static org.apache.hudi.common.util.StringUtils.EMPTY_STRING;
/**
- * Performs rollback using marker files generated during the write..
+ * Performs rollback using marker files generated during the writes.
*/
public class MarkerBasedRollbackStrategy<T, I, K, O> implements BaseRollbackPlanActionExecutor.RollbackStrategy {
private static final Logger LOG = LoggerFactory.getLogger(MarkerBasedRollbackStrategy.class);
protected final HoodieTable<?, ?, ?, ?> table;
-
protected final transient HoodieEngineContext context;
-
protected final HoodieWriteConfig config;
-
protected final String basePath;
-
protected final String instantTime;
public MarkerBasedRollbackStrategy(HoodieTable<?, ?, ?, ?> table, HoodieEngineContext context, HoodieWriteConfig config, String instantTime) {
this.table = table;
this.context = context;
- this.basePath = table.getMetaClient().getBasePath();
+ this.basePath = table.getMetaClient().getBasePathV2().toString();
this.config = config;
this.instantTime = instantTime;
}
@@ -78,22 +77,28 @@ public class MarkerBasedRollbackStrategy<T, I, K, O> implements BaseRollbackPlan
return context.map(markerPaths, markerFilePath -> {
String typeStr = markerFilePath.substring(markerFilePath.lastIndexOf(".") + 1);
IOType type = IOType.valueOf(typeStr);
+ String fileNameWithPartitionToRollback = WriteMarkers.stripMarkerSuffix(markerFilePath);
+ Path fullFilePathToRollback = new Path(basePath, fileNameWithPartitionToRollback);
+ String partitionPath = FSUtils.getRelativePartitionPath(new Path(basePath), fullFilePathToRollback.getParent());
switch (type) {
case MERGE:
case CREATE:
- String fileToDelete = WriteMarkers.stripMarkerSuffix(markerFilePath);
- Path fullDeletePath = new Path(basePath, fileToDelete);
- String partitionPath = FSUtils.getRelativePartitionPath(new Path(basePath), fullDeletePath.getParent());
- return new HoodieRollbackRequest(partitionPath, EMPTY_STRING, EMPTY_STRING,
- Collections.singletonList(fullDeletePath.toString()),
+ String fileId = null;
+ String baseInstantTime = null;
+ if (FSUtils.isBaseFile(fullFilePathToRollback)) {
+ HoodieBaseFile baseFileToDelete = new HoodieBaseFile(fullFilePathToRollback.toString());
+ fileId = baseFileToDelete.getFileId();
+ baseInstantTime = baseFileToDelete.getCommitTime();
+ } else if (FSUtils.isLogFile(fullFilePathToRollback)) {
+ throw new HoodieRollbackException("Log files should have only APPEND as IOTypes " + fullFilePathToRollback);
+ }
+ Objects.requireNonNull(fileId, "Cannot find valid fileId from path: " + fullFilePathToRollback);
+ Objects.requireNonNull(baseInstantTime, "Cannot find valid base instant from path: " + fullFilePathToRollback);
+ return new HoodieRollbackRequest(partitionPath, fileId, baseInstantTime,
+ Collections.singletonList(fullFilePathToRollback.toString()),
Collections.emptyMap());
case APPEND:
- // NOTE: This marker file-path does NOT correspond to a log-file, but rather is a phony
- // path serving as a "container" for the following components:
- // - Base file's file-id
- // - Base file's commit instant
- // - Partition path
- return getRollbackRequestForAppend(instantToRollback, WriteMarkers.stripMarkerSuffix(markerFilePath));
+ return getRollbackRequestForAppend(instantToRollback, fileNameWithPartitionToRollback);
default:
throw new HoodieRollbackException("Unknown marker type, during rollback of " + instantToRollback);
}
@@ -103,36 +108,51 @@ public class MarkerBasedRollbackStrategy<T, I, K, O> implements BaseRollbackPlan
}
}
- protected HoodieRollbackRequest getRollbackRequestForAppend(HoodieInstant instantToRollback, String markerFilePath) throws IOException {
- Path baseFilePathForAppend = new Path(basePath, markerFilePath);
- String fileId = FSUtils.getFileIdFromFilePath(baseFilePathForAppend);
- String baseCommitTime = FSUtils.getCommitTime(baseFilePathForAppend.getName());
- String relativePartitionPath = FSUtils.getRelativePartitionPath(new Path(basePath), baseFilePathForAppend.getParent());
- Path partitionPath = FSUtils.getPartitionPath(config.getBasePath(), relativePartitionPath);
-
- // NOTE: Since we're rolling back incomplete Delta Commit, it only could have appended its
- // block to the latest log-file
- // TODO(HUDI-1517) use provided marker-file's path instead
- Option<HoodieLogFile> latestLogFileOption = FSUtils.getLatestLogFile(table.getMetaClient().getFs(), partitionPath, fileId,
- HoodieFileFormat.HOODIE_LOG.getFileExtension(), baseCommitTime);
-
- // Log file can be deleted if the commit to rollback is also the commit that created the fileGroup
- if (latestLogFileOption.isPresent() && baseCommitTime.equals(instantToRollback.getTimestamp())) {
- Path fullDeletePath = new Path(partitionPath, latestLogFileOption.get().getFileName());
- return new HoodieRollbackRequest(relativePartitionPath, EMPTY_STRING, EMPTY_STRING,
- Collections.singletonList(fullDeletePath.toString()),
- Collections.emptyMap());
- }
-
- Map<String, Long> logFilesWithBlocsToRollback = new HashMap<>();
- if (latestLogFileOption.isPresent()) {
- HoodieLogFile latestLogFile = latestLogFileOption.get();
- // NOTE: Marker's don't carry information about the cumulative size of the blocks that have been appended,
- // therefore we simply stub this value.
- logFilesWithBlocsToRollback = Collections.singletonMap(latestLogFile.getFileStatus().getPath().toString(), -1L);
+ protected HoodieRollbackRequest getRollbackRequestForAppend(HoodieInstant instantToRollback, String fileNameWithPartitionToRollback) {
+ Path fullLogFilePath = new Path(basePath, fileNameWithPartitionToRollback);
+ String relativePartitionPath = FSUtils.getRelativePartitionPath(new Path(basePath), fullLogFilePath.getParent());
+ String fileId;
+ String baseCommitTime;
+ Option<HoodieLogFile> latestLogFileOption;
+ Map<String, Long> logBlocksToBeDeleted = new HashMap<>();
+ // Old marker files may be generated from base file name before HUDI-1517. keep compatible with them.
+ if (FSUtils.isBaseFile(fullLogFilePath)) {
+ LOG.warn("Find old marker type for log file: " + fileNameWithPartitionToRollback);
+ fileId = FSUtils.getFileIdFromFilePath(fullLogFilePath);
+ baseCommitTime = FSUtils.getCommitTime(fullLogFilePath.getName());
+ Path partitionPath = FSUtils.getPartitionPath(config.getBasePath(), relativePartitionPath);
+
+ // NOTE: Since we're rolling back incomplete Delta Commit, it only could have appended its
+ // block to the latest log-file
+ try {
+ latestLogFileOption = FSUtils.getLatestLogFile(table.getMetaClient().getFs(), partitionPath, fileId,
+ HoodieFileFormat.HOODIE_LOG.getFileExtension(), baseCommitTime);
+ if (latestLogFileOption.isPresent() && baseCommitTime.equals(instantToRollback.getTimestamp())) {
+ Path fullDeletePath = new Path(partitionPath, latestLogFileOption.get().getFileName());
+ return new HoodieRollbackRequest(relativePartitionPath, EMPTY_STRING, EMPTY_STRING,
+ Collections.singletonList(fullDeletePath.toString()),
+ Collections.emptyMap());
+ }
+ if (latestLogFileOption.isPresent()) {
+ HoodieLogFile latestLogFile = latestLogFileOption.get();
+ // NOTE: Markers don't carry information about the cumulative size of the blocks that have been appended,
+ // therefore we simply stub this value.
+ logBlocksToBeDeleted = Collections.singletonMap(latestLogFile.getFileStatus().getPath().toString(), latestLogFile.getFileStatus().getLen());
+ }
+ return new HoodieRollbackRequest(relativePartitionPath, fileId, baseCommitTime, Collections.emptyList(), logBlocksToBeDeleted);
+ } catch (IOException ioException) {
+ throw new HoodieIOException(
+ "Failed to get latestLogFile for fileId: " + fileId + " in partition: " + partitionPath,
+ ioException);
+ }
+ } else {
+ HoodieLogFile logFileToRollback = new HoodieLogFile(fullLogFilePath);
+ fileId = logFileToRollback.getFileId();
+ baseCommitTime = logFileToRollback.getBaseCommitTime();
+ // NOTE: We don't strictly need the exact size, but this size needs to be positive to pass metadata payload validation.
+ // Therefore, we simply stub this value (1L), instead of doing a fs call to get the exact size.
+ logBlocksToBeDeleted = Collections.singletonMap(logFileToRollback.getPath().getName(), 1L);
}
-
- return new HoodieRollbackRequest(relativePartitionPath, fileId, baseCommitTime, Collections.emptyList(),
- logFilesWithBlocsToRollback);
+ return new HoodieRollbackRequest(relativePartitionPath, fileId, baseCommitTime, Collections.emptyList(), logBlocksToBeDeleted);
}
}
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 c3ee30ed3f4..c804bd1933f 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
@@ -19,6 +19,7 @@
package org.apache.hudi.table.action.rollback;
import org.apache.hudi.avro.model.HoodieRollbackPlan;
+import org.apache.hudi.avro.model.HoodieRollbackRequest;
import org.apache.hudi.common.HoodieRollbackStat;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.log.block.HoodieCommandBlock;
@@ -94,4 +95,18 @@ public class RollbackUtils {
return new HoodieRollbackStat(stat1.getPartitionPath(), successDeleteFiles, failedDeleteFiles, commandBlocksCount, logFilesFromFailedCommit);
}
+ static HoodieRollbackRequest mergeRollbackRequest(HoodieRollbackRequest rollbackRequest1, HoodieRollbackRequest rollbackRequest2) {
+ checkArgument(rollbackRequest1.getPartitionPath().equals(rollbackRequest2.getPartitionPath()));
+ checkArgument((rollbackRequest1.getFileId().equals(rollbackRequest2.getFileId())));
+ checkArgument((rollbackRequest1.getLatestBaseInstant().equals(rollbackRequest2.getLatestBaseInstant())));
+ final List<String> filesToBeDeleted = new ArrayList<>();
+ final Map<String, Long> logBlocksToBeDeleted = new HashMap<>();
+ Option.ofNullable(rollbackRequest1.getFilesToBeDeleted()).ifPresent(filesToBeDeleted::addAll);
+ Option.ofNullable(rollbackRequest1.getLogBlocksToBeDeleted()).ifPresent(logBlocksToBeDeleted::putAll);
+ Option.ofNullable(rollbackRequest2.getFilesToBeDeleted()).ifPresent(filesToBeDeleted::addAll);
+ Option.ofNullable(rollbackRequest2.getLogBlocksToBeDeleted()).ifPresent(logBlocksToBeDeleted::putAll);
+
+ return new HoodieRollbackRequest(rollbackRequest1.getPartitionPath(), rollbackRequest1.getFileId(), rollbackRequest1.getLatestBaseInstant(),
+ filesToBeDeleted, logBlocksToBeDeleted);
+ }
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/DirectWriteMarkers.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/DirectWriteMarkers.java
index a540c21a8a7..abe1c63d576 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/DirectWriteMarkers.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/DirectWriteMarkers.java
@@ -47,8 +47,11 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
+import java.util.LinkedList;
import java.util.List;
+import java.util.Queue;
import java.util.Set;
+import java.util.function.Predicate;
import static org.apache.hudi.table.marker.ConflictDetectionUtils.getDefaultEarlyConflictDetectionStrategy;
@@ -59,6 +62,9 @@ import static org.apache.hudi.table.marker.ConflictDetectionUtils.getDefaultEarl
public class DirectWriteMarkers extends WriteMarkers {
private static final Logger LOG = LoggerFactory.getLogger(DirectWriteMarkers.class);
+ private static final Predicate<String> APPEND_MARKER_PREDICATE = pathStr -> pathStr.contains(HoodieTableMetaClient.MARKER_EXTN) && pathStr.endsWith(IOType.APPEND.name());
+ private static final Predicate<String> NOT_APPEND_MARKER_PREDICATE = pathStr -> pathStr.contains(HoodieTableMetaClient.MARKER_EXTN) && !pathStr.endsWith(IOType.APPEND.name());
+
private final transient FileSystem fs;
public DirectWriteMarkers(FileSystem fs, String basePath, String markerFolderPath, String instantTime) {
@@ -76,7 +82,7 @@ public class DirectWriteMarkers extends WriteMarkers {
/**
* Deletes Marker directory corresponding to an instant.
*
- * @param context HoodieEngineContext.
+ * @param context HoodieEngineContext.
* @param parallelism parallelism for deletion.
*/
public boolean deleteMarkerDir(HoodieEngineContext context, int parallelism) {
@@ -94,20 +100,7 @@ public class DirectWriteMarkers extends WriteMarkers {
@Override
public Set<String> createdAndMergedDataPaths(HoodieEngineContext context, int parallelism) throws IOException {
Set<String> dataFiles = new HashSet<>();
-
- FileStatus[] topLevelStatuses = fs.listStatus(markerDirPath);
- List<String> subDirectories = new ArrayList<>();
- for (FileStatus topLevelStatus: topLevelStatuses) {
- if (topLevelStatus.isFile()) {
- String pathStr = topLevelStatus.getPath().toString();
- if (pathStr.contains(HoodieTableMetaClient.MARKER_EXTN) && !pathStr.endsWith(IOType.APPEND.name())) {
- dataFiles.add(translateMarkerToDataPath(pathStr));
- }
- } else {
- subDirectories.add(topLevelStatus.getPath().toString());
- }
- }
-
+ List<String> subDirectories = getSubDirectoriesByMarkerCondition(fs.listStatus(markerDirPath), dataFiles, NOT_APPEND_MARKER_PREDICATE);
if (subDirectories.size() > 0) {
parallelism = Math.min(subDirectories.size(), parallelism);
SerializableConfiguration serializedConf = new SerializableConfiguration(fs.getConf());
@@ -120,7 +113,7 @@ public class DirectWriteMarkers extends WriteMarkers {
while (itr.hasNext()) {
FileStatus status = itr.next();
String pathStr = status.getPath().toString();
- if (pathStr.contains(HoodieTableMetaClient.MARKER_EXTN) && !pathStr.endsWith(IOType.APPEND.name())) {
+ if (NOT_APPEND_MARKER_PREDICATE.test(pathStr)) {
result.add(translateMarkerToDataPath(pathStr));
}
}
@@ -131,6 +124,56 @@ public class DirectWriteMarkers extends WriteMarkers {
return dataFiles;
}
+ public Set<String> getAppendedLogPaths(HoodieEngineContext context, int parallelism) throws IOException {
+ Set<String> logFiles = new HashSet<>();
+ List<String> subDirectories = getSubDirectoriesByMarkerCondition(fs.listStatus(markerDirPath), logFiles, APPEND_MARKER_PREDICATE);
+
+ if (subDirectories.size() > 0) {
+ parallelism = Math.min(subDirectories.size(), parallelism);
+ SerializableConfiguration serializedConf = new SerializableConfiguration(fs.getConf());
+ context.setJobStatus(this.getClass().getSimpleName(), "Obtaining marker files for all created, merged paths");
+ logFiles.addAll(context.flatMap(subDirectories, directory -> {
+ Queue<Path> candidatesDirs = new LinkedList<>();
+ candidatesDirs.add(new Path(directory));
+ List<String> result = new ArrayList<>();
+ while (!candidatesDirs.isEmpty()) {
+ Path path = candidatesDirs.remove();
+ FileSystem fileSystem = HadoopFSUtils.getFs(path, serializedConf.get());
+ RemoteIterator<FileStatus> itr = fileSystem.listStatusIterator(path);
+ while (itr.hasNext()) {
+ FileStatus status = itr.next();
+ if (status.isDirectory()) {
+ candidatesDirs.add(status.getPath());
+ } else {
+ String pathStr = status.getPath().toString();
+ if (APPEND_MARKER_PREDICATE.test(pathStr)) {
+ result.add(translateMarkerToDataPath(pathStr));
+ }
+ }
+ }
+ }
+ return result.stream();
+ }, parallelism));
+ }
+
+ return logFiles;
+ }
+
+ private List<String> getSubDirectoriesByMarkerCondition(FileStatus[] topLevelStatuses, Set<String> dataFiles, Predicate<String> pathCondition) {
+ List<String> subDirectories = new ArrayList<>();
+ for (FileStatus topLevelStatus : topLevelStatuses) {
+ if (topLevelStatus.isFile()) {
+ String pathStr = topLevelStatus.getPath().toString();
+ if (pathCondition.test(pathStr)) {
+ dataFiles.add(translateMarkerToDataPath(pathStr));
+ }
+ } else {
+ subDirectories.add(topLevelStatus.getPath().toString());
+ }
+ }
+ return subDirectories;
+ }
+
private String translateMarkerToDataPath(String markerPath) {
String rPath = MarkerUtils.stripMarkerFolderPrefix(markerPath, basePath, instantTime);
return stripMarkerSuffix(rPath);
@@ -159,8 +202,8 @@ public class DirectWriteMarkers extends WriteMarkers {
}
@Override
- protected Option<Path> create(String partitionPath, String dataFileName, IOType type, boolean checkIfExists) {
- return create(getMarkerPath(partitionPath, dataFileName, type), checkIfExists);
+ protected Option<Path> create(String partitionPath, String fileName, IOType type, boolean checkIfExists) {
+ return create(getMarkerPath(partitionPath, fileName, type), checkIfExists);
}
@Override
@@ -200,7 +243,7 @@ public class DirectWriteMarkers extends WriteMarkers {
} catch (IOException e) {
throw new HoodieException("Failed to create marker file " + markerPath, e);
}
- LOG.info("[direct] Created marker file " + markerPath.toString()
+ LOG.info("[direct] Created marker file " + markerPath
+ " in " + timer.endTimer() + " ms");
return Option.of(markerPath);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java
index 427af12c6c4..1eae90c8225 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java
@@ -48,6 +48,7 @@ import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.hudi.common.table.marker.MarkerOperation.ALL_MARKERS_URL;
+import static org.apache.hudi.common.table.marker.MarkerOperation.APPEND_MARKERS_URL;
import static org.apache.hudi.common.table.marker.MarkerOperation.CREATE_AND_MERGE_MARKERS_URL;
import static org.apache.hudi.common.table.marker.MarkerOperation.CREATE_MARKER_URL;
import static org.apache.hudi.common.table.marker.MarkerOperation.DELETE_MARKER_DIR_URL;
@@ -123,6 +124,19 @@ public class TimelineServerBasedWriteMarkers extends WriteMarkers {
}
}
+ @Override
+ public Set<String> getAppendedLogPaths(HoodieEngineContext context, int parallelism) throws IOException {
+ Map<String, String> paramsMap = Collections.singletonMap(MARKER_DIR_PATH_PARAM, markerDirPath.toString());
+ try {
+ Set<String> markerPaths = executeRequestToTimelineServer(
+ APPEND_MARKERS_URL, paramsMap, new TypeReference<Set<String>>() {}, RequestMethod.GET);
+ return markerPaths.stream().map(WriteMarkers::stripMarkerSuffix).collect(Collectors.toSet());
+ } catch (IOException e) {
+ throw new HoodieRemoteException("Failed to get APPEND log file paths in "
+ + markerDirPath.toString(), e);
+ }
+ }
+
@Override
public Set<String> allMarkerFilePaths() {
Map<String, String> paramsMap = Collections.singletonMap(MARKER_DIR_PATH_PARAM, markerDirPath.toString());
@@ -135,9 +149,9 @@ public class TimelineServerBasedWriteMarkers extends WriteMarkers {
}
@Override
- protected Option<Path> create(String partitionPath, String dataFileName, IOType type, boolean checkIfExists) {
+ protected Option<Path> create(String partitionPath, String fileName, IOType type, boolean checkIfExists) {
HoodieTimer timer = HoodieTimer.start();
- String markerFileName = getMarkerFileName(dataFileName, type);
+ String markerFileName = getMarkerFileName(fileName, type);
Map<String, String> paramsMap = getConfigMap(partitionPath, markerFileName, false);
boolean success = executeCreateMarkerRequest(paramsMap, partitionPath, markerFileName);
@@ -151,10 +165,10 @@ public class TimelineServerBasedWriteMarkers extends WriteMarkers {
}
@Override
- public Option<Path> createWithEarlyConflictDetection(String partitionPath, String dataFileName, IOType type, boolean checkIfExists,
+ public Option<Path> createWithEarlyConflictDetection(String partitionPath, String fileName, IOType type, boolean checkIfExists,
HoodieWriteConfig config, String fileId, HoodieActiveTimeline activeTimeline) {
HoodieTimer timer = new HoodieTimer().startTimer();
- String markerFileName = getMarkerFileName(dataFileName, type);
+ String markerFileName = getMarkerFileName(fileName, type);
Map<String, String> paramsMap = getConfigMap(partitionPath, markerFileName, true);
boolean success = executeCreateMarkerRequest(paramsMap, partitionPath, markerFileName);
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkers.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkers.java
index 93aba9c0f89..01c8c99618a 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkers.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkers.java
@@ -59,12 +59,12 @@ public abstract class WriteMarkers implements Serializable {
* Creates a marker without checking if the marker already exists.
*
* @param partitionPath partition path in the table.
- * @param dataFileName data file name.
+ * @param fileName file name.
* @param type write IO type.
* @return the marker path.
*/
- public Option<Path> create(String partitionPath, String dataFileName, IOType type) {
- return create(partitionPath, dataFileName, type, false);
+ public Option<Path> create(String partitionPath, String fileName, IOType type) {
+ return create(partitionPath, fileName, type, false);
}
/**
@@ -72,14 +72,14 @@ public abstract class WriteMarkers implements Serializable {
* This can invoke marker-based early conflict detection when enabled for multi-writers.
*
* @param partitionPath partition path in the table
- * @param dataFileName data file name
+ * @param fileName file name
* @param type write IO type
* @param writeConfig Hudi write configs.
* @param fileId File ID.
* @param activeTimeline Active timeline for the write operation.
* @return the marker path.
*/
- public Option<Path> create(String partitionPath, String dataFileName, IOType type, HoodieWriteConfig writeConfig,
+ public Option<Path> create(String partitionPath, String fileName, IOType type, HoodieWriteConfig writeConfig,
String fileId, HoodieActiveTimeline activeTimeline) {
if (writeConfig.getWriteConcurrencyMode().supportsOptimisticConcurrencyControl()
&& writeConfig.isEarlyConflictDetectionEnable()) {
@@ -88,23 +88,23 @@ public abstract class WriteMarkers implements Serializable {
// TODO If current is compact or clustering then create marker directly without early conflict detection.
// Need to support early conflict detection between table service and common writers.
if (pendingCompactionTimeline.containsInstant(instantTime) || pendingReplaceTimeline.containsInstant(instantTime)) {
- return create(partitionPath, dataFileName, type, false);
+ return create(partitionPath, fileName, type, false);
}
- return createWithEarlyConflictDetection(partitionPath, dataFileName, type, false, writeConfig, fileId, activeTimeline);
+ return createWithEarlyConflictDetection(partitionPath, fileName, type, false, writeConfig, fileId, activeTimeline);
}
- return create(partitionPath, dataFileName, type, false);
+ return create(partitionPath, fileName, type, false);
}
/**
* Creates a marker if the marker does not exist.
*
* @param partitionPath partition path in the table
- * @param dataFileName data file name
+ * @param fileName file name
* @param type write IO type
* @return the marker path or empty option if already exists
*/
- public Option<Path> createIfNotExists(String partitionPath, String dataFileName, IOType type) {
- return create(partitionPath, dataFileName, type, true);
+ public Option<Path> createIfNotExists(String partitionPath, String fileName, IOType type) {
+ return create(partitionPath, fileName, type, true);
}
/**
@@ -161,27 +161,27 @@ public abstract class WriteMarkers implements Serializable {
}
/**
- * Gets the marker file name, in the format of "[data_file_name].marker.[IO_type]".
+ * Gets the marker file name, in the format of "[file_name].marker.[IO_type]".
*
- * @param dataFileName data file name
+ * @param fileName file name
* @param type IO type
* @return the marker file name
*/
- protected String getMarkerFileName(String dataFileName, IOType type) {
- return String.format("%s%s.%s", dataFileName, HoodieTableMetaClient.MARKER_EXTN, type.name());
+ protected static String getMarkerFileName(String fileName, IOType type) {
+ return String.format("%s%s.%s", fileName, HoodieTableMetaClient.MARKER_EXTN, type.name());
}
/**
* Returns the marker path. Would create the partition path first if not exists
*
* @param partitionPath The partition path
- * @param dataFileName The data file name
+ * @param fileName The file name
* @param type The IO type
* @return path of the marker file
*/
- protected Path getMarkerPath(String partitionPath, String dataFileName, IOType type) {
+ protected Path getMarkerPath(String partitionPath, String fileName, IOType type) {
Path path = FSUtils.getPartitionPath(markerDirPath, partitionPath);
- String markerFileName = getMarkerFileName(dataFileName, type);
+ String markerFileName = getMarkerFileName(fileName, type);
return new Path(path, markerFileName);
}
@@ -203,11 +203,19 @@ public abstract class WriteMarkers implements Serializable {
/**
* @param context {@code HoodieEngineContext} instance.
* @param parallelism parallelism for reading the marker files in the directory.
- * @return all the data file paths of write IO type "CREATE" and "MERGE"
+ * @return all the data file or log file paths of write IO type "CREATE" and "MERGE"
* @throws IOException
*/
public abstract Set<String> createdAndMergedDataPaths(HoodieEngineContext context, int parallelism) throws IOException;
+ /**
+ * @param context {@code HoodieEngineContext} instance.
+ * @param parallelism parallelism for reading the marker files in the directory.
+ * @return all the log file paths of write IO type "APPEND"
+ * @throws IOException
+ */
+ public abstract Set<String> getAppendedLogPaths(HoodieEngineContext context, int parallelism) throws IOException;
+
/**
* @return all the marker paths
* @throws IOException
@@ -218,19 +226,19 @@ public abstract class WriteMarkers implements Serializable {
* Creates a marker.
*
* @param partitionPath partition path in the table
- * @param dataFileName data file name
+ * @param fileName file name
* @param type write IO type
* @param checkIfExists whether to check if the marker already exists
* @return the marker path or empty option if already exists and {@code checkIfExists} is true
*/
- abstract Option<Path> create(String partitionPath, String dataFileName, IOType type, boolean checkIfExists);
+ abstract Option<Path> create(String partitionPath, String fileName, IOType type, boolean checkIfExists);
/**
* Creates a marker with early conflict detection for multi-writers. If conflict is detected,
* an exception is thrown to fail the write operation.
*
* @param partitionPath partition path in the table.
- * @param dataFileName data file name.
+ * @param fileName file name.
* @param type write IO type.
* @param checkIfExists whether to check if the marker already exists.
* @param config Hudi write configs.
@@ -238,6 +246,6 @@ public abstract class WriteMarkers implements Serializable {
* @param activeTimeline Active timeline for the write operation.
* @return the marker path or empty option if already exists and {@code checkIfExists} is true.
*/
- public abstract Option<Path> createWithEarlyConflictDetection(String partitionPath, String dataFileName, IOType type, boolean checkIfExists,
+ public abstract Option<Path> createWithEarlyConflictDetection(String partitionPath, String fileName, IOType type, boolean checkIfExists,
HoodieWriteConfig config, String fileId, HoodieActiveTimeline activeTimeline);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/ZeroToOneUpgradeHandler.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/ZeroToOneUpgradeHandler.java
index 9096c4e05cd..6f5a7e69e27 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/ZeroToOneUpgradeHandler.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/ZeroToOneUpgradeHandler.java
@@ -118,8 +118,9 @@ public class ZeroToOneUpgradeHandler implements UpgradeHandler {
List<HoodieRollbackRequest> hoodieRollbackRequests =
new ListingBasedRollbackStrategy(table, context, table.getConfig(), commitInstantOpt.get().getTimestamp(), false)
.getRollbackRequests(commitInstantOpt.get());
- return new BaseRollbackHelper(table.getMetaClient(), table.getConfig())
- .collectRollbackStats(context, commitInstantOpt.get(), hoodieRollbackRequests);
+ String rollbackInstantTime = HoodieActiveTimeline.createNewInstantTime();
+ return new BaseRollbackHelper(table, table.getConfig())
+ .collectRollbackStats(context, rollbackInstantTime, commitInstantOpt.get(), hoodieRollbackRequests);
}
/**
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/utils/TestCommitMetadataUtils.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/utils/TestCommitMetadataUtils.java
new file mode 100644
index 00000000000..6d1d038ff9f
--- /dev/null
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/utils/TestCommitMetadataUtils.java
@@ -0,0 +1,177 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.client.utils;
+
+import org.apache.hudi.common.config.SerializableConfiguration;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieDeltaWriteStat;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.marker.MarkerType;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
+import org.apache.hudi.common.testutils.FileCreateUtils;
+import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
+import org.apache.hudi.common.testutils.HoodieTestTable;
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.table.marker.WriteMarkers;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+
+import static org.apache.hudi.HoodieTestCommitGenerator.getBaseFilename;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TestCommitMetadataUtils extends HoodieCommonTestHarness {
+
+ private final HoodieWriteConfig writeConfig = mock(HoodieWriteConfig.class);
+ private final HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class);
+ private final HoodieWrapperFileSystem fileSystem = mock(HoodieWrapperFileSystem.class);
+ private final HoodieEngineContext context = mock(HoodieEngineContext.class);
+ private final HoodieTable table = mock(HoodieTable.class);
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ initPath();
+ initMetaClient();
+ }
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ cleanMetaClient();
+ }
+
+ @Test
+ public void testReconcileMetadataForMissingFiles() throws IOException {
+ // Mock table type as MERGE_ON_READ and action as DELTA_COMMIT
+ when(table.getMetaClient()).thenReturn(metaClient);
+ Mockito.when(table.getConfig()).thenReturn(writeConfig);
+ when(metaClient.getTableType()).thenReturn(HoodieTableType.MERGE_ON_READ);
+ when(metaClient.getFs()).thenReturn(fileSystem);
+ when(metaClient.getBasePath()).thenReturn(basePath);
+ when(metaClient.getMarkerFolderPath(any())).thenReturn(basePath + ".hoodie/.temp");
+ when(table.getContext()).thenReturn(context);
+ when(context.getHadoopConf()).thenReturn(new SerializableConfiguration(new Configuration()));
+ when(writeConfig.getViewStorageConfig()).thenReturn(FileSystemViewStorageConfig.newBuilder().build());
+ when(writeConfig.getMarkersType()).thenReturn(MarkerType.DIRECT);
+ when(writeConfig.getBasePath()).thenReturn(basePath);
+ String commitActionType = HoodieActiveTimeline.DELTA_COMMIT_ACTION;
+ String instantTime = HoodieActiveTimeline.createNewInstantTime();
+
+ // Setup dummy commit metadata
+ String p0 = "2020/01/01";
+ String p1 = "2020/01/02";
+ String file1P0C0 = UUID.randomUUID().toString();
+ String file1P1C0 = UUID.randomUUID().toString();
+ Map<String, List<String>> part1ToFileId = Collections.unmodifiableMap(new HashMap<String, List<String>>() {
+ {
+ put(p0, CollectionUtils.createImmutableList(file1P0C0));
+ put(p1, CollectionUtils.createImmutableList(file1P1C0));
+ }
+ });
+ Pair<HoodieCommitMetadata, List<String>> commitMetadataWithLogFiles =
+ generateCommitMetadata(instantTime, part1ToFileId, basePath, 1, 2);
+
+ // Assume these are paths to log files that were supposed to be in commitMetadata but are missing
+ Set<String> missingLogFiles = new HashSet<>(Arrays.asList("path/to/log1", "path/to/log2"));
+ // Mocking the behavior to return missing log files
+ WriteMarkers markers = mock(WriteMarkers.class);
+ // Add valid log files along with missing ones
+ when(markers.getAppendedLogPaths(any(), anyInt())).thenReturn(missingLogFiles);
+ when(table.getFileSystemView()).thenReturn(mock(org.apache.hudi.common.table.view.HoodieTableFileSystemView.class));
+ missingLogFiles.addAll(commitMetadataWithLogFiles.getRight());
+ when(markers.getAppendedLogPaths(any(), anyInt())).thenReturn(missingLogFiles);
+ when(table.getFileSystemView()).thenReturn(mock(org.apache.hudi.common.table.view.HoodieTableFileSystemView.class));
+
+ // Mock filesystem and file status
+ FileSystem fs = mock(FileSystem.class);
+ Configuration hadoopConf = new Configuration();
+ when(table.getHadoopConf()).thenReturn(hadoopConf);
+ when(fs.exists(any())).thenReturn(true);
+
+ // Call the method under test
+ HoodieCommitMetadata reconciledMetadata = CommitMetadataUtils.reconcileMetadataForMissingFiles(
+ table, commitActionType, instantTime, commitMetadataWithLogFiles.getLeft(), writeConfig, context, hadoopConf, this.getClass().getSimpleName());
+
+ // Assertions to verify if the missing files are added
+ assertFalse(reconciledMetadata.getPartitionToWriteStats().isEmpty(), "CommitMetadata should not be empty after reconciliation");
+ assertEquals(2, reconciledMetadata.getPartitionToWriteStats().size());
+ assertTrue(reconciledMetadata.getPartitionToWriteStats().containsKey(p0), "Partition " + p0 + " should be present in the commit metadata");
+ assertTrue(reconciledMetadata.getPartitionToWriteStats().containsKey(p1), "Partition " + p1 + " should be present in the commit metadata");
+ assertEquals(1, reconciledMetadata.getPartitionToWriteStats().get(p0).size(), "There should be 1 write stats for partition " + p0);
+ assertEquals(1, reconciledMetadata.getPartitionToWriteStats().get(p1).size(), "There should be 1 write stats for partition " + p1);
+ assertEquals(file1P0C0, reconciledMetadata.getPartitionToWriteStats().get(p0).get(0).getFileId(), "FileId for partition " + p0 + " should be " + file1P0C0);
+ assertEquals(file1P1C0, reconciledMetadata.getPartitionToWriteStats().get(p1).get(0).getFileId(), "FileId for partition " + p1 + " should be " + file1P1C0);
+ }
+
+ private static Pair<HoodieCommitMetadata, List<String>> generateCommitMetadata(String instantTime, Map<String, List<String>> partitionToFilePaths,
+ String basePath, int... versions) {
+ HoodieCommitMetadata metadata = new HoodieCommitMetadata();
+ metadata.addMetadata(HoodieCommitMetadata.SCHEMA_KEY, HoodieTestTable.PHONY_TABLE_SCHEMA);
+ List<String> allLogFiles = new ArrayList<>();
+ partitionToFilePaths.forEach((partitionPath, fileList) -> fileList.forEach(f -> {
+ HoodieDeltaWriteStat writeStat = new HoodieDeltaWriteStat();
+ List<String> logFiles = new ArrayList<>();
+ for (int version : versions) {
+ try {
+ logFiles.add(FileCreateUtils.createLogFile(basePath, partitionPath, instantTime, f, version));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ allLogFiles.addAll(logFiles);
+ writeStat.setPartitionPath(partitionPath);
+ writeStat.setPath(partitionPath + "/" + getBaseFilename(instantTime, f));
+ writeStat.setFileId(f);
+ writeStat.setTotalWriteBytes(1);
+ writeStat.setFileSizeInBytes(1);
+ writeStat.setTotalLogBlocks(logFiles.size());
+ writeStat.setLogFiles(logFiles);
+ metadata.addWriteStat(partitionPath, writeStat);
+ }));
+ return Pair.of(metadata, allLogFiles);
+ }
+}
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/providers/HoodieMetaClientProvider.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/providers/HoodieMetaClientProvider.java
index 0cd7ed5a715..721cc5e7c5b 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/providers/HoodieMetaClientProvider.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/providers/HoodieMetaClientProvider.java
@@ -22,6 +22,9 @@ package org.apache.hudi.testutils.providers;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.table.view.SyncableFileSystemView;
+import org.apache.hudi.common.testutils.HoodieTestTable;
+import org.apache.hudi.exception.HoodieIOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
@@ -37,4 +40,15 @@ public interface HoodieMetaClientProvider {
HoodieTableMetaClient metaClient, HoodieTimeline visibleActiveTimeline, FileStatus[] fileStatuses) {
return new HoodieTableFileSystemView(metaClient, visibleActiveTimeline, fileStatuses);
}
+
+ default SyncableFileSystemView getFileSystemViewWithUnCommittedSlices(HoodieTableMetaClient metaClient) {
+ try {
+ return new HoodieTableFileSystemView(metaClient,
+ metaClient.getActiveTimeline(),
+ HoodieTestTable.of(metaClient).listAllBaseAndLogFiles()
+ );
+ } catch (IOException ioe) {
+ throw new HoodieIOException("Error getting file system view", ioe);
+ }
+ }
}
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkAppendHandle.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkAppendHandle.java
index 3dc76ed435e..5bd0c26aed3 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkAppendHandle.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkAppendHandle.java
@@ -20,7 +20,10 @@ package org.apache.hudi.io;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.engine.TaskContextSupplier;
+import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.IOType;
+import org.apache.hudi.common.table.log.HoodieLogFileWriteCallback;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.marker.WriteMarkers;
@@ -49,7 +52,6 @@ public class FlinkAppendHandle<T, I, K, O>
private static final Logger LOG = LoggerFactory.getLogger(FlinkAppendHandle.class);
private boolean isClosed = false;
- private final WriteMarkers writeMarkers;
public FlinkAppendHandle(
HoodieWriteConfig config,
@@ -60,17 +62,22 @@ public class FlinkAppendHandle<T, I, K, O>
Iterator<HoodieRecord<T>> recordItr,
TaskContextSupplier taskContextSupplier) {
super(config, instantTime, hoodieTable, partitionPath, fileId, recordItr, taskContextSupplier);
- this.writeMarkers = WriteMarkersFactory.get(config.getMarkersType(), hoodieTable, instantTime);
}
- @Override
- protected void createMarkerFile(String partitionPath, String dataFileName) {
- // In some rare cases, the task was pulled up again with same write file name,
- // for e.g, reuse the small log files from last commit instant.
-
- // Just skip the marker creation if it already exists, the new data would append to
- // the file directly.
- writeMarkers.createIfNotExists(partitionPath, dataFileName, getIOType());
+ protected HoodieLogFileWriteCallback getLogWriteCallback() {
+ return new AppendLogWriteCallback() {
+ @Override
+ public boolean preLogFileOpen(HoodieLogFile logFileToAppend) {
+ // In some rare cases, the task was pulled up again with same write file name,
+ // for e.g, reuse the small log files from last commit instant.
+
+ // Just skip the marker creation if it already exists, the new data would append to
+ // the file directly.
+ WriteMarkers writeMarkers = WriteMarkersFactory.get(config.getMarkersType(), hoodieTable, instantTime);
+ writeMarkers.createIfNotExists(partitionPath, logFileToAppend.getFileName(), IOType.APPEND);
+ return true;
+ }
+ };
}
@Override
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
index 6fdfee16bbe..4ec886e1edb 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
@@ -20,6 +20,7 @@ package org.apache.hudi.client;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.client.embedded.EmbeddedTimelineService;
+import org.apache.hudi.client.utils.CommitMetadataUtils;
import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodieData.HoodieDataCacheKey;
import org.apache.hudi.common.engine.HoodieEngineContext;
@@ -34,6 +35,7 @@ import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.data.HoodieJavaRDD;
+import org.apache.hudi.exception.HoodieCommitException;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
import org.apache.hudi.index.HoodieIndex;
@@ -55,6 +57,7 @@ import org.apache.spark.api.java.JavaSparkContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -104,6 +107,17 @@ public class SparkRDDWriteClient<T> extends
return commitStats(instantTime, HoodieJavaRDD.of(writeStatuses), writeStats, extraMetadata, commitActionType, partitionToReplacedFileIds, extraPreCommitFunc);
}
+ @Override
+ protected HoodieCommitMetadata reconcileCommitMetadata(HoodieTable table, String commitActionType, String instantTime, HoodieCommitMetadata originalMetadata) {
+ try {
+ return CommitMetadataUtils.reconcileMetadataForMissingFiles(table, commitActionType,
+ instantTime, originalMetadata, config, context, hadoopConf, this.getClass().getSimpleName());
+ } catch (IOException e) {
+ throw new HoodieCommitException("Failed to fix commit metadata for spurious log files "
+ + config.getBasePath() + " at time " + instantTime, e);
+ }
+ }
+
@Override
protected HoodieTable createTable(HoodieWriteConfig config, Configuration hadoopConf) {
return HoodieSparkTable.create(config, context);
@@ -183,7 +197,7 @@ public class SparkRDDWriteClient<T> extends
/**
* Removes all existing records from the partitions affected and inserts the given HoodieRecords, into the table.
*
- * @param records HoodieRecords to insert
+ * @param records HoodieRecords to insert
* @param instantTime Instant time of the commit
* @return JavaRDD[WriteStatus] - RDD of WriteStatus to inspect errors and counts
*/
@@ -199,7 +213,7 @@ public class SparkRDDWriteClient<T> extends
/**
* Removes all existing records of the Hoodie table and inserts the given HoodieRecords, into the table.
*
- * @param records HoodieRecords to insert
+ * @param records HoodieRecords to insert
* @param instantTime Instant time of the commit
* @return JavaRDD[WriteStatus] - RDD of WriteStatus to inspect errors and counts
*/
@@ -252,7 +266,7 @@ public class SparkRDDWriteClient<T> extends
public JavaRDD<WriteStatus> deletePrepped(JavaRDD<HoodieRecord<T>> preppedRecord, String instantTime) {
HoodieTable<T, HoodieData<HoodieRecord<T>>, HoodieData<HoodieKey>, HoodieData<WriteStatus>> table = initTable(WriteOperationType.DELETE_PREPPED, Option.ofNullable(instantTime));
preWrite(instantTime, WriteOperationType.DELETE_PREPPED, table.getMetaClient());
- HoodieWriteMetadata<HoodieData<WriteStatus>> result = table.deletePrepped(context,instantTime, HoodieJavaRDD.of(preppedRecord));
+ HoodieWriteMetadata<HoodieData<WriteStatus>> result = table.deletePrepped(context, instantTime, HoodieJavaRDD.of(preppedRecord));
HoodieWriteMetadata<JavaRDD<WriteStatus>> resultRDD = result.clone(HoodieJavaRDD.getJavaRDD(result.getWriteStatuses()));
return postWrite(resultRDD, instantTime, table);
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/data/HoodieJavaPairRDD.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/data/HoodieJavaPairRDD.java
index 9019fb43ff0..3c7c014dbf1 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/data/HoodieJavaPairRDD.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/data/HoodieJavaPairRDD.java
@@ -142,6 +142,14 @@ public class HoodieJavaPairRDD<K, V> implements HoodiePairData<K, V> {
new ImmutablePair<>(tuple._2._1, Option.ofNullable(tuple._2._2.orElse(null)))))));
}
+ @Override
+ public <W> HoodiePairData<K, Pair<V, W>> join(HoodiePairData<K, W> other) {
+ return HoodieJavaPairRDD.of(JavaPairRDD.fromJavaRDD(
+ pairRDDData.join(HoodieJavaPairRDD.getJavaPairRDD(other))
+ .map(tuple -> new Tuple2<>(tuple._1,
+ new ImmutablePair<>(tuple._2._1, tuple._2._2)))));
+ }
+
@Override
public List<Pair<K, V>> collectAsList() {
return pairRDDData.map(t -> Pair.of(t._1, t._2)).collect();
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
index 0ca910fd721..36a167e32f5 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
@@ -307,6 +307,7 @@ public abstract class BaseSparkCommitActionExecutor<T> extends
try {
HoodieActiveTimeline activeTimeline = table.getActiveTimeline();
HoodieCommitMetadata metadata = result.getCommitMetadata().get();
+ metadata = appendMetadataForMissingFiles(metadata);
writeTableMetadata(metadata, result.getWriteStatuses(), actionType);
activeTimeline.saveAsComplete(new HoodieInstant(true, getCommitActionType(), instantTime),
Option.of(getUTF8Bytes(metadata.toJsonString())));
@@ -318,6 +319,10 @@ public abstract class BaseSparkCommitActionExecutor<T> extends
}
}
+ protected HoodieCommitMetadata appendMetadataForMissingFiles(HoodieCommitMetadata commitMetadata) throws IOException {
+ return commitMetadata;
+ }
+
protected Map<String, List<String>> getPartitionToReplacedFileIds(HoodieWriteMetadata<HoodieData<WriteStatus>> writeStatuses) {
return Collections.emptyMap();
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/BaseSparkDeltaCommitActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/BaseSparkDeltaCommitActionExecutor.java
index be69be05c84..793baccbacd 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/BaseSparkDeltaCommitActionExecutor.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/BaseSparkDeltaCommitActionExecutor.java
@@ -20,6 +20,8 @@ package org.apache.hudi.table.action.deltacommit;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.client.utils.CommitMetadataUtils;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.util.Option;
@@ -50,16 +52,21 @@ public abstract class BaseSparkDeltaCommitActionExecutor<T>
private SparkUpsertDeltaCommitPartitioner<T> mergeOnReadUpsertPartitioner;
public BaseSparkDeltaCommitActionExecutor(HoodieSparkEngineContext context, HoodieWriteConfig config, HoodieTable table,
- String instantTime, WriteOperationType operationType) {
+ String instantTime, WriteOperationType operationType) {
this(context, config, table, instantTime, operationType, Option.empty());
}
public BaseSparkDeltaCommitActionExecutor(HoodieSparkEngineContext context, HoodieWriteConfig config, HoodieTable table,
- String instantTime, WriteOperationType operationType,
- Option<Map<String, String>> extraMetadata) {
+ String instantTime, WriteOperationType operationType,
+ Option<Map<String, String>> extraMetadata) {
super(context, config, table, instantTime, operationType, extraMetadata);
}
+ @Override
+ protected HoodieCommitMetadata appendMetadataForMissingFiles(HoodieCommitMetadata commitMetadata) throws IOException {
+ return CommitMetadataUtils.reconcileMetadataForMissingFiles(table, getCommitActionType(), instantTime, commitMetadata, config, context, hadoopConf, this.getClass().getSimpleName());
+ }
+
@Override
public Partitioner getUpsertPartitioner(WorkloadProfile profile) {
if (profile == null) {
@@ -71,7 +78,7 @@ public abstract class BaseSparkDeltaCommitActionExecutor<T>
@Override
public Iterator<List<WriteStatus>> handleUpdate(String partitionPath, String fileId,
- Iterator<HoodieRecord<T>> recordItr) throws IOException {
+ Iterator<HoodieRecord<T>> recordItr) throws IOException {
LOG.info("Merging updates for commit " + instantTime + " for file " + fileId);
if (!table.getIndex().canIndexLogFiles() && mergeOnReadUpsertPartitioner != null
&& mergeOnReadUpsertPartitioner.getSmallFileIds().contains(fileId)) {
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
index c554e99e7e8..dc0e78e229e 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieBackedMetadata.java
@@ -3374,6 +3374,64 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
HoodieTimer timer = HoodieTimer.start();
HoodieSparkEngineContext engineContext = new HoodieSparkEngineContext(jsc);
+ validateMetadata(config, ignoreFilesWithCommit, fs, basePath, metaClient, hadoopConf, engineContext, tableMetadata);
+
+ HoodieBackedTableMetadataWriter<JavaRDD<HoodieRecord>> metadataWriter = metadataWriter(client);
+ assertNotNull(metadataWriter, "MetadataWriter should have been initialized");
+
+ // Validate write config for metadata table
+ HoodieWriteConfig metadataWriteConfig = metadataWriter.getWriteConfig();
+ assertFalse(metadataWriteConfig.isMetadataTableEnabled(), "No metadata table for metadata table");
+
+ // Metadata table should be in sync with the dataset
+ HoodieTableMetaClient metadataMetaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(metadataTableBasePath).build();
+
+ // Metadata table is MOR
+ assertEquals(metadataMetaClient.getTableType(), HoodieTableType.MERGE_ON_READ, "Metadata Table should be MOR");
+
+ // Metadata table is HFile format
+ assertEquals(metadataMetaClient.getTableConfig().getBaseFileFormat(), HoodieFileFormat.HFILE,
+ "Metadata Table base file format should be HFile");
+
+ // Metadata table has a fixed number of partitions
+ // Cannot use FSUtils.getAllFoldersWithPartitionMetaFile for this as that function filters all directory
+ // in the .hoodie folder.
+ List<String> metadataTablePartitions = FSUtils.getAllPartitionPaths(engineContext, getMetadataTableBasePath(basePath),
+ false, false);
+ assertEquals(metadataWriter.getEnabledPartitionTypes().size(), metadataTablePartitions.size());
+
+ final Map<String, MetadataPartitionType> metadataEnabledPartitionTypes = new HashMap<>();
+ metadataWriter.getEnabledPartitionTypes().forEach(e -> metadataEnabledPartitionTypes.put(e.getPartitionPath(), e));
+
+ // Metadata table should automatically compact and clean
+ // versions are +1 as autoclean / compaction happens end of commits
+ int numFileVersions = metadataWriteConfig.getCleanerFileVersionsRetained() + 1;
+ HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(metadataMetaClient, metadataMetaClient.getActiveTimeline());
+ metadataTablePartitions.forEach(partition -> {
+ List<FileSlice> latestSlices = fsView.getLatestFileSlices(partition).collect(Collectors.toList());
+ assertTrue(latestSlices.stream().map(FileSlice::getBaseFile).count() <= latestSlices.size(), "Should have a single latest base file per file group");
+ List<HoodieLogFile> logFiles = latestSlices.get(0).getLogFiles().collect(Collectors.toList());
+ try {
+ if (FILES.getPartitionPath().equals(partition)) {
+ HoodieTable table = HoodieSparkTable.create(config, engineContext);
+ verifyMetadataRawRecords(table, logFiles, false);
+ }
+ if (COLUMN_STATS.getPartitionPath().equals(partition)) {
+ verifyMetadataColumnStatsRecords(logFiles);
+ }
+ } catch (IOException e) {
+ LOG.error("Metadata record validation failed", e);
+ fail("Metadata record validation failed");
+ }
+ });
+
+ // TODO: include validation for record_index partition here.
+ LOG.info("Validation time=" + timer.endTimer());
+ }
+
+ public static void validateMetadata(HoodieWriteConfig config, Option<String> ignoreFilesWithCommit,
+ FileSystem fs, String basePath, HoodieTableMetaClient metaClient,
+ Configuration hadoopConf, HoodieSparkEngineContext engineContext, HoodieTableMetadata tableMetadata) throws IOException {
// Partitions should match
FileSystemBackedTableMetadata fsBackedTableMetadata = new FileSystemBackedTableMetadata(engineContext, metaClient.getTableConfig(),
@@ -3417,6 +3475,8 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
Collections.sort(fsFileNames);
Collections.sort(metadataFilenames);
+ fsFileNames.forEach(n -> System.out.println("FSFILENAME: " + n));
+ metadataFilenames.forEach(n -> System.out.println("METADATAFILENAME: " + n));
assertEquals(fsStatuses.length, partitionToFilesMap.get(partitionPath.toString()).length);
// File sizes should be valid
@@ -3466,57 +3526,6 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
assertTrue(false, "Exception should not be raised: " + e);
}
});
-
- try (HoodieBackedTableMetadataWriter<JavaRDD<HoodieRecord>> metadataWriter = metadataWriter(client)) {
- assertNotNull(metadataWriter, "MetadataWriter should have been initialized");
-
- // Validate write config for metadata table
- HoodieWriteConfig metadataWriteConfig = metadataWriter.getWriteConfig();
- assertFalse(metadataWriteConfig.isMetadataTableEnabled(), "No metadata table for metadata table");
-
- // Metadata table should be in sync with the dataset
- HoodieTableMetaClient metadataMetaClient = HoodieTableMetaClient.builder().setConf(hadoopConf).setBasePath(metadataTableBasePath).build();
-
- // Metadata table is MOR
- assertEquals(metadataMetaClient.getTableType(), HoodieTableType.MERGE_ON_READ, "Metadata Table should be MOR");
-
- // Metadata table is HFile format
- assertEquals(metadataMetaClient.getTableConfig().getBaseFileFormat(), HoodieFileFormat.HFILE,
- "Metadata Table base file format should be HFile");
-
- // Metadata table has a fixed number of partitions
- // Cannot use FSUtils.getAllFoldersWithPartitionMetaFile for this as that function filters all directory
- // in the .hoodie folder.
- List<String> metadataTablePartitions = FSUtils.getAllPartitionPaths(engineContext, getMetadataTableBasePath(basePath), false, false);
- assertEquals(metadataWriter.getEnabledPartitionTypes().size(), metadataTablePartitions.size());
-
- final Map<String, MetadataPartitionType> metadataEnabledPartitionTypes = new HashMap<>();
- metadataWriter.getEnabledPartitionTypes().forEach(e -> metadataEnabledPartitionTypes.put(e.getPartitionPath(), e));
-
- // Metadata table should automatically compact and clean
- // versions are +1 as autoclean / compaction happens end of commits
- int numFileVersions = metadataWriteConfig.getCleanerFileVersionsRetained() + 1;
- HoodieTableFileSystemView fsView = new HoodieTableFileSystemView(metadataMetaClient, metadataMetaClient.getActiveTimeline());
- metadataTablePartitions.forEach(partition -> {
- List<FileSlice> latestSlices = fsView.getLatestFileSlices(partition).collect(Collectors.toList());
- assertTrue(latestSlices.stream().map(FileSlice::getBaseFile).count() <= latestSlices.size(), "Should have a single latest base file per file group");
- List<HoodieLogFile> logFiles = latestSlices.get(0).getLogFiles().collect(Collectors.toList());
- try {
- if (FILES.getPartitionPath().equals(partition)) {
- verifyMetadataRawRecords(table, logFiles, false);
- }
- if (COLUMN_STATS.getPartitionPath().equals(partition)) {
- verifyMetadataColumnStatsRecords(logFiles);
- }
- } catch (IOException e) {
- LOG.error("Metadata record validation failed", e);
- fail("Metadata record validation failed");
- }
- });
-
- // TODO: include validation for record_index partition here.
- LOG.info("Validation time=" + timer.endTimer());
- }
}
private void verifyMetadataColumnStatsRecords(List<HoodieLogFile> logFiles) throws IOException {
@@ -3572,7 +3581,7 @@ public class TestHoodieBackedMetadata extends TestHoodieMetadataBase {
.create(hadoopConf, client.getConfig(), new HoodieSparkEngineContext(jsc));
}
- private HoodieTableMetadata metadata(SparkRDDWriteClient client) {
+ public static HoodieTableMetadata metadata(SparkRDDWriteClient client) {
HoodieWriteConfig clientConfig = client.getConfig();
return HoodieTableMetadata.create(client.getEngineContext(), clientConfig.getMetadataConfig(), clientConfig.getBasePath());
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/data/TestHoodieJavaPairRDD.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/data/TestHoodieJavaPairRDD.java
new file mode 100644
index 00000000000..75bc888a71d
--- /dev/null
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/data/TestHoodieJavaPairRDD.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.data;
+
+import org.apache.hudi.common.data.HoodiePairData;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+
+import org.apache.spark.SparkConf;
+import org.apache.spark.api.java.JavaPairRDD;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+import scala.Tuple2;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@SuppressWarnings("unchecked")
+public class TestHoodieJavaPairRDD {
+
+ private static JavaSparkContext jsc;
+
+ @BeforeEach
+ public void setUp() {
+ // Initialize Spark context and JavaPairRDD mock
+ SparkConf conf = new SparkConf().setAppName("HoodieJavaPairRDDJoinTest").setMaster("local[2]");
+ jsc = new JavaSparkContext(conf);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (jsc != null) {
+ jsc.stop();
+ }
+ }
+
+ @Test
+ public void testJoinOperation() {
+ JavaPairRDD<String, String> partitionRecordKeyPairRDD = jsc.parallelize(Arrays.asList(
+ new Tuple2<>("2017/10/22", "003"),
+ new Tuple2<>("2017/10/22", "002"),
+ new Tuple2<>("2017/10/22", "005"),
+ new Tuple2<>("2017/10/22", "004"))).mapToPair(t -> t);
+
+ JavaPairRDD<String, String> otherPairRDD = jsc.parallelize(Arrays.asList(
+ new Tuple2<>("2017/10/22", "value1"),
+ new Tuple2<>("2017/10/22", "value2"))).mapToPair(t -> t);
+
+ HoodieJavaPairRDD<String, String> hoodiePairData = HoodieJavaPairRDD.of(partitionRecordKeyPairRDD);
+ HoodieJavaPairRDD<String, String> otherHoodiePairData = HoodieJavaPairRDD.of(otherPairRDD);
+
+ HoodiePairData<String, Pair<String, String>> result = hoodiePairData.join(otherHoodiePairData);
+
+ List<Pair<String, Pair<String, String>>> resultList = result.collectAsList();
+ assertEquals(8, resultList.size());
+ resultList.forEach(item -> {
+ assertEquals("2017/10/22", item.getLeft());
+ assertTrue(Arrays.asList("003", "002", "005", "004").contains(item.getRight().getLeft()));
+ assertTrue(Arrays.asList("value1", "value2").contains(item.getRight().getRight()));
+ });
+ }
+
+ @Test
+ public void testLeftOuterJoinOperation() {
+ JavaPairRDD<String, String> partitionRecordKeyPairRDD = jsc.parallelize(Arrays.asList(
+ new Tuple2<>("2017/10/22", "003"),
+ new Tuple2<>("2017/10/22", "002"),
+ new Tuple2<>("2017/10/22", "005"),
+ new Tuple2<>("2017/10/22", "004"))).mapToPair(t -> t);
+
+ JavaPairRDD<String, String> otherPairRDD = jsc.parallelize(Arrays.asList(
+ new Tuple2<>("2017/10/22", "value1"))).mapToPair(t -> t);
+
+ HoodieJavaPairRDD<String, String> hoodiePairData = HoodieJavaPairRDD.of(partitionRecordKeyPairRDD);
+ HoodieJavaPairRDD<String, String> otherHoodiePairData = HoodieJavaPairRDD.of(otherPairRDD);
+
+ HoodiePairData<String, Pair<String, Option<String>>> result = hoodiePairData.leftOuterJoin(otherHoodiePairData);
+
+ List<Pair<String, Pair<String, Option<String>>>> resultList = result.collectAsList();
+ assertEquals(4, resultList.size());
+ resultList.forEach(item -> {
+ assertEquals("2017/10/22", item.getLeft());
+ assertTrue(Arrays.asList("003", "002", "005", "004").contains(item.getRight().getLeft()));
+ assertEquals(Option.of("value1"), item.getRight().getRight());
+ });
+ }
+}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
index b2fab0ae492..2188d7246fa 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
@@ -416,7 +416,7 @@ public class TestHoodieMergeOnReadTable extends SparkClientFunctionalTestHarness
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testMetadataStatsOnCommit(Boolean rollbackUsingMarkers) throws Exception {
- HoodieWriteConfig cfg = getConfigBuilder(false, rollbackUsingMarkers, IndexType.INMEMORY)
+ HoodieWriteConfig cfg = getConfigBuilder(false, rollbackUsingMarkers, IndexType.BLOOM)
.withAvroSchemaValidate(false)
.withAllowAutoEvolutionColumnDrop(true)
.withAutoCommit(false)
@@ -463,7 +463,6 @@ public class TestHoodieMergeOnReadTable extends SparkClientFunctionalTestHarness
records = dataGen.generateUpdates(instantTime, records);
writeRecords = jsc().parallelize(records, 1);
statuses = client.upsert(writeRecords, instantTime);
- //assertTrue(client.commit(instantTime, statuses), "Commit should succeed");
inserts = 0;
int upserts = 0;
List<WriteStatus> writeStatusList = statuses.collect();
@@ -476,6 +475,11 @@ public class TestHoodieMergeOnReadTable extends SparkClientFunctionalTestHarness
assertEquals(0, inserts);
assertEquals(200, upserts);
+ if (!rollbackUsingMarkers) {
+ // we can do listing based rollback only when commit is completed
+ assertTrue(client.commit(instantTime, statuses), "Commit should succeed");
+ }
+
client.rollback(instantTime);
// Read from commit file
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestCopyOnWriteRollbackActionExecutor.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestCopyOnWriteRollbackActionExecutor.java
index ca881308fc5..a6c43f0974c 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestCopyOnWriteRollbackActionExecutor.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestCopyOnWriteRollbackActionExecutor.java
@@ -103,7 +103,7 @@ public class TestCopyOnWriteRollbackActionExecutor extends HoodieClientRollbackT
.withBaseFilesInPartition(p1, "id21").getLeft()
.withBaseFilesInPartition(p2, "id22").getLeft();
- HoodieWriteConfig writeConfig = getConfigBuilder().withRollbackUsingMarkers(false).build();
+ HoodieWriteConfig writeConfig = getConfigBuilder().withRollbackUsingMarkers(false).withEmbeddedTimelineServerEnabled(false).build();
HoodieTable table = this.getHoodieTable(metaClient, writeConfig);
HoodieInstant needRollBackInstant = new HoodieInstant(false, HoodieTimeline.COMMIT_ACTION, "002");
String rollbackInstant = "003";
@@ -261,7 +261,7 @@ public class TestCopyOnWriteRollbackActionExecutor extends HoodieClientRollbackT
.addCommit("003")
.withBaseFilesInPartition(p3, fileLengths);
- HoodieTable table = this.getHoodieTable(metaClient, getConfigBuilder().withRollbackUsingMarkers(false).build());
+ HoodieTable table = this.getHoodieTable(metaClient, getConfigBuilder().withRollbackUsingMarkers(false).withEmbeddedTimelineServerEnabled(false).build());
HoodieInstant needRollBackInstant = new HoodieInstant(false, HoodieTimeline.COMMIT_ACTION, "003");
// Schedule rollback
@@ -352,7 +352,7 @@ public class TestCopyOnWriteRollbackActionExecutor extends HoodieClientRollbackT
.withBaseFilesInPartition(p1, "id21").getLeft()
.withBaseFilesInPartition(p2, "id22").getLeft();
- HoodieTable table = this.getHoodieTable(metaClient, getConfigBuilder().withRollbackBackupEnabled(true).build());
+ HoodieTable table = this.getHoodieTable(metaClient, getConfigBuilder().withRollbackBackupEnabled(true).withEmbeddedTimelineServerEnabled(false).build());
HoodieInstant needRollBackInstant = new HoodieInstant(false, HoodieTimeline.COMMIT_ACTION, "002");
// Create the rollback plan and perform the rollback
@@ -411,7 +411,7 @@ public class TestCopyOnWriteRollbackActionExecutor extends HoodieClientRollbackT
public void testRollbackWhenReplaceCommitIsPresent() throws Exception {
// insert data
- HoodieWriteConfig writeConfig = getConfigBuilder().withAutoCommit(false).build();
+ HoodieWriteConfig writeConfig = getConfigBuilder().withAutoCommit(false).withEmbeddedTimelineServerEnabled(false).build();
SparkRDDWriteClient writeClient = getHoodieWriteClient(writeConfig);
// Create a base commit.
@@ -444,7 +444,7 @@ public class TestCopyOnWriteRollbackActionExecutor extends HoodieClientRollbackT
// Now execute clustering on the saved instant and do not allow it to commit.
ClusteringTestUtils.runClusteringOnInstant(clusteringClient, false, false, clusteringInstant1);
- HoodieTable table = this.getHoodieTable(metaClient, getConfigBuilder().build());
+ HoodieTable table = this.getHoodieTable(metaClient, getConfigBuilder().withEmbeddedTimelineServerEnabled(false).build());
HoodieInstant needRollBackInstant = new HoodieInstant(false, HoodieTimeline.COMMIT_ACTION, secondCommit);
// Schedule rollback
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMergeOnReadRollbackActionExecutor.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMergeOnReadRollbackActionExecutor.java
index 426f7e489d4..02a9ed977bf 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMergeOnReadRollbackActionExecutor.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestMergeOnReadRollbackActionExecutor.java
@@ -323,7 +323,7 @@ public class TestMergeOnReadRollbackActionExecutor extends HoodieClientRollbackT
assertEquals(2, hoodieWriteStatOptionList.get(0).getNumInserts());
// Rollback
- HoodieInstant rollBackInstant = new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, "002");
+ HoodieInstant rollBackInstant = new HoodieInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION, "002");
BaseRollbackPlanActionExecutor mergeOnReadRollbackPlanActionExecutor =
new BaseRollbackPlanActionExecutor(context, cfg, table, "003", rollBackInstant, false,
cfg.shouldRollbackUsingMarkers(), false);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestRollbackUtils.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestRollbackUtils.java
index c22a2aef424..fa479bb9683 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestRollbackUtils.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestRollbackUtils.java
@@ -18,6 +18,7 @@
package org.apache.hudi.table.action.rollback;
+import org.apache.hudi.avro.model.HoodieRollbackRequest;
import org.apache.hudi.common.HoodieRollbackStat;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
@@ -30,14 +31,17 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertIterableEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestRollbackUtils {
private static final String BASE_FILE_EXTENSION = HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().getFileExtension();
@@ -121,4 +125,38 @@ public class TestRollbackUtils {
assertEquals(Collections.singletonMap(generateFileStatus(partitionPath1 + "dataFile1.log"), 10L),
dataFilesOnlyStatMerge2.getCommandBlocksCount());
}
+
+ @Test
+ public void testMergeRollbackRequestSuccess() {
+ String partitionPath = "partition/path";
+ String fileId = "fileId";
+ String latestBaseInstant = "latestBaseInstant";
+ List<String> filesToBeDeleted1 = Arrays.asList("file1", "file2");
+ Map<String, Long> logBlocksToBeDeleted1 = new HashMap<>();
+ logBlocksToBeDeleted1.put("block1", 1L);
+
+ List<String> filesToBeDeleted2 = Arrays.asList("file3", "file4");
+ Map<String, Long> logBlocksToBeDeleted2 = new HashMap<>();
+ logBlocksToBeDeleted2.put("block2", 2L);
+
+ HoodieRollbackRequest request1 = new HoodieRollbackRequest(partitionPath, fileId, latestBaseInstant, filesToBeDeleted1, logBlocksToBeDeleted1);
+ HoodieRollbackRequest request2 = new HoodieRollbackRequest(partitionPath, fileId, latestBaseInstant, filesToBeDeleted2, logBlocksToBeDeleted2);
+
+ HoodieRollbackRequest mergedRequest = RollbackUtils.mergeRollbackRequest(request1, request2);
+
+ // Verify
+ assertEquals(partitionPath, mergedRequest.getPartitionPath());
+ assertEquals(fileId, mergedRequest.getFileId());
+ assertEquals(latestBaseInstant, mergedRequest.getLatestBaseInstant());
+ assertTrue(mergedRequest.getFilesToBeDeleted().containsAll(Arrays.asList("file1", "file2", "file3", "file4")));
+ assertEquals(2, mergedRequest.getLogBlocksToBeDeleted().size());
+ assertTrue(mergedRequest.getLogBlocksToBeDeleted().keySet().containsAll(Arrays.asList("block1", "block2")));
+ }
+
+ @Test
+ public void testMergeRollbackRequestWithMismatchArguments() {
+ HoodieRollbackRequest request1 = new HoodieRollbackRequest("partition/path", "fileId", "latestBaseInstant", null, null);
+ HoodieRollbackRequest request2 = new HoodieRollbackRequest("partition/path2", "fileId2", "latestBaseInstant2", null, null);
+ assertThrows(IllegalArgumentException.class, () -> RollbackUtils.mergeRollbackRequest(request1, request2));
+ }
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkCopyOnWriteTableRollback.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkCopyOnWriteTableRollback.java
new file mode 100644
index 00000000000..9f3af5651b1
--- /dev/null
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkCopyOnWriteTableRollback.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.table.functional;
+
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+
+import static org.apache.hudi.common.model.HoodieTableType.COPY_ON_WRITE;
+
+@Tag("functional")
+public class TestHoodieSparkCopyOnWriteTableRollback extends TestHoodieSparkRollback {
+
+ /**
+ * Scenario: data table is updated, no changes to MDT
+ */
+ @Test
+ public void testRollbackWithFailurePreMDT() throws IOException {
+ testRollbackWithFailurePreMDT(COPY_ON_WRITE);
+ }
+
+ /**
+ * Scenario: data table is updated, deltacommit is completed in MDT
+ */
+ @Test
+ public void testRollbackWithFailurePostMDT() throws IOException {
+ testRollbackWithFailurePostMDT(COPY_ON_WRITE);
+ }
+
+ /**
+ * Scenario: data table is updated, deltacommit is completed in MDT then during rollback,
+ * data table is updated, no changes to MDT
+ */
+ @Test
+ public void testRollbackWithFailurePostMDTRollbackFailsPreMDT() throws IOException {
+ testRollbackWithFailurePostMDT(COPY_ON_WRITE, true);
+ }
+
+ /**
+ * Scenario: data table is updated, deltacommit of interest is inflight in MDT
+ */
+ @Test
+ public void testRollbackWithFailureInMDT() throws Exception {
+ testRollbackWithFailureinMDT(COPY_ON_WRITE);
+ }
+
+}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableInsertUpdateDelete.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableInsertUpdateDelete.java
index 73d551b0ae0..84165f274a3 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableInsertUpdateDelete.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableInsertUpdateDelete.java
@@ -21,18 +21,26 @@ package org.apache.hudi.table.functional;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieBaseFile;
+import org.apache.hudi.common.model.HoodieCommitMetadata;
+import org.apache.hudi.common.model.HoodieDeltaWriteStat;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.log.block.HoodieAvroDataBlock;
+import org.apache.hudi.common.table.log.block.HoodieDataBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
+import org.apache.hudi.common.table.view.SyncableFileSystemView;
import org.apache.hudi.common.table.view.TableFileSystemView;
import org.apache.hudi.common.testutils.FileCreateUtils;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
@@ -43,12 +51,15 @@ import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.table.HoodieSparkTable;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.HoodieWriteMetadata;
+import org.apache.hudi.table.marker.WriteMarkers;
+import org.apache.hudi.table.marker.WriteMarkersFactory;
import org.apache.hudi.testutils.HoodieClientTestUtils;
import org.apache.hudi.testutils.HoodieMergeOnReadTestUtils;
import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;
import org.apache.spark.api.java.JavaRDD;
@@ -60,8 +71,11 @@ import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
+import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -344,18 +358,48 @@ public class TestHoodieSparkMergeOnReadTableInsertUpdateDelete extends SparkClie
List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 100);
JavaRDD<HoodieRecord> recordsRDD = jsc().parallelize(records, 1);
JavaRDD<WriteStatus> statuses = writeClient.insert(recordsRDD, newCommitTime);
+ long expectedLogFileNum = statuses.map(writeStatus -> (HoodieDeltaWriteStat) writeStatus.getStat())
+ .flatMap(deltaWriteStat -> deltaWriteStat.getLogFiles().iterator())
+ .count();
+ // inject a fake log file to test marker file for log file
+ HoodieDeltaWriteStat correctWriteStat = (HoodieDeltaWriteStat) statuses.map(WriteStatus::getStat).take(1).get(0);
+ assertTrue(FSUtils.isLogFile(new Path(correctWriteStat.getPath())));
+ HoodieLogFile correctLogFile = new HoodieLogFile(correctWriteStat.getPath());
+ String correctWriteToken = FSUtils.getWriteTokenFromLogPath(correctLogFile.getPath());
+
+ final String newToken = generateNewDifferentWriteToken(correctWriteToken);
+ String originalLogfileName = correctLogFile.getPath().getName();
+ String logFileWithoutWriteToken = originalLogfileName.substring(0, originalLogfileName.lastIndexOf("_") + 1);
+ String newLogFileName = logFileWithoutWriteToken + newToken;
+ Path parentPath = correctLogFile.getPath().getParent();
+ FileSystem fs = parentPath.getFileSystem(jsc().hadoopConfiguration());
+ // copy to create another log file w/ diff write token.
+ fs.copyToLocalFile(new Path(config.getBasePath(), correctLogFile.getPath().toString()), new Path(config.getBasePath().toString() + "/" + parentPath, newLogFileName));
+
+ // generate marker for the same
+ final WriteMarkers writeMarkers = WriteMarkersFactory.get(config.getMarkersType(),
+ HoodieSparkTable.create(config, context()), newCommitTime);
+ writeMarkers.create(correctWriteStat.getPartitionPath(), newLogFileName, IOType.APPEND);
+
+ // check marker for additional log generated
+ assertTrue(writeMarkers.allMarkerFilePaths().stream().anyMatch(marker -> marker.contains(newToken)));
+ SyncableFileSystemView unCommittedFsView = getFileSystemViewWithUnCommittedSlices(metaClient);
+ // check additional log generated
+ assertTrue(unCommittedFsView.getAllFileSlices(correctWriteStat.getPartitionPath())
+ .flatMap(FileSlice::getLogFiles).map(HoodieLogFile::getPath)
+ .anyMatch(path -> path.getName().equals(newLogFileName)));
writeClient.commit(newCommitTime, statuses);
HoodieTable table = HoodieSparkTable.create(config, context(), metaClient);
table.getHoodieView().sync();
TableFileSystemView.SliceView tableRTFileSystemView = table.getSliceView();
-
+ // get log file number from filesystem view
long numLogFiles = 0;
for (String partitionPath : dataGen.getPartitionPaths()) {
List<FileSlice> allSlices = tableRTFileSystemView.getLatestFileSlices(partitionPath).collect(Collectors.toList());
assertEquals(0, allSlices.stream().filter(fileSlice -> fileSlice.getBaseFile().isPresent()).count());
assertTrue(allSlices.stream().anyMatch(fileSlice -> fileSlice.getLogFiles().count() > 0));
- long logFileCount = allSlices.stream().filter(fileSlice -> fileSlice.getLogFiles().count() > 0).count();
+ long logFileCount = allSlices.stream().mapToLong(fileSlice -> fileSlice.getLogFiles().count()).sum();
if (logFileCount > 0) {
// check the log versions start from the base version
assertTrue(allSlices.stream().map(slice -> slice.getLogFiles().findFirst().get().getLogVersion())
@@ -363,16 +407,35 @@ public class TestHoodieSparkMergeOnReadTableInsertUpdateDelete extends SparkClie
}
numLogFiles += logFileCount;
}
-
- assertTrue(numLogFiles > 0);
+ // check log file number in file system to cover all log files including additional log files created with spark task retries
+ assertEquals(expectedLogFileNum + 1, numLogFiles);
+ Option<byte[]> bytes = table.getActiveTimeline().getInstantDetails(table.getActiveTimeline().getDeltaCommitTimeline().lastInstant().get());
+ // check log file number in commit metadata cover all log files mentioned above
+ HoodieCommitMetadata commitMetadata = HoodieCommitMetadata.fromBytes(bytes.get(), HoodieCommitMetadata.class);
+ assertEquals(expectedLogFileNum + 1, commitMetadata.getWriteStats().size());
// Do a compaction
String instantTime = writeClient.scheduleCompaction(Option.empty()).get().toString();
HoodieWriteMetadata<JavaRDD<WriteStatus>> compactionMetadata = writeClient.compact(instantTime);
String extension = table.getBaseFileExtension();
Collection<List<HoodieWriteStat>> stats = compactionMetadata.getCommitMetadata().get().getPartitionToWriteStats().values();
- assertEquals(numLogFiles, stats.stream().flatMap(Collection::stream).filter(state -> state.getPath().contains(extension)).count());
- assertEquals(numLogFiles, stats.stream().mapToLong(Collection::size).sum());
+ assertEquals(3, stats.stream().flatMap(Collection::stream).filter(state -> state.getPath().contains(extension)).count());
writeClient.commitCompaction(instantTime, compactionMetadata.getCommitMetadata().get(), Option.empty());
}
}
+
+ private HoodieDataBlock getLogBlock(List<HoodieRecord> hoodieRecords, String schema) {
+ Map<HoodieLogBlock.HeaderMetadataType, String> header = new HashMap<>();
+ header.put(HoodieLogBlock.HeaderMetadataType.INSTANT_TIME, "100");
+ header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, schema);
+ return new HoodieAvroDataBlock(hoodieRecords, header, HoodieRecord.RECORD_KEY_METADATA_FIELD);
+ }
+
+ private String generateNewDifferentWriteToken(String correctWriteToken) {
+ Random random = new Random();
+ String fakeToken = "";
+ do {
+ fakeToken = Math.abs(random.nextInt()) + "-" + Math.abs(random.nextInt()) + "-" + Math.abs(random.nextInt());
+ } while (fakeToken.equals(correctWriteToken));
+ return fakeToken;
+ }
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
index e492682fef3..ab976d10b6b 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
@@ -19,8 +19,12 @@
package org.apache.hudi.table.functional;
+import org.apache.hudi.avro.model.HoodieRollbackMetadata;
+import org.apache.hudi.avro.model.HoodieRollbackPartitionMetadata;
+import org.apache.hudi.avro.model.HoodieRollbackPlan;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.HoodieRollbackStat;
import org.apache.hudi.common.config.HoodieStorageConfig;
import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieBaseFile;
@@ -34,6 +38,7 @@ import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.marker.MarkerType;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
@@ -41,7 +46,6 @@ import org.apache.hudi.common.table.view.HoodieTableFileSystemView;
import org.apache.hudi.common.table.view.SyncableFileSystemView;
import org.apache.hudi.common.table.view.TableFileSystemView;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
-import org.apache.hudi.common.testutils.HoodieTestTable;
import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
@@ -53,8 +57,8 @@ import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.table.HoodieSparkTable;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.HoodieWriteMetadata;
+import org.apache.hudi.table.action.rollback.MergeOnReadRollbackActionExecutor;
import org.apache.hudi.testutils.HoodieMergeOnReadTestUtils;
-import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.fs.FileStatus;
@@ -63,6 +67,8 @@ import org.apache.spark.api.java.JavaRDD;
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;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
@@ -78,16 +84,19 @@ import java.util.Properties;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static org.apache.hudi.common.model.HoodieTableType.MERGE_ON_READ;
+import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.NO_PARTITION_PATH;
import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
import static org.junit.jupiter.api.Assertions.assertAll;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@Tag("functional")
-public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunctionalTestHarness {
+public class TestHoodieSparkMergeOnReadTableRollback extends TestHoodieSparkRollback {
@ParameterizedTest
@ValueSource(booleans = {true, false})
@@ -131,7 +140,7 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
assertNoWriteErrors(statuses);
// Set TableType to MOR
- metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ);
+ metaClient = getHoodieMetaClient(MERGE_ON_READ);
// rollback a COW commit when TableType is MOR
client.rollback(newCommitTime);
@@ -158,7 +167,7 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
Properties properties = CollectionUtils.copy(cfg.getProps());
properties.setProperty(HoodieTableConfig.BASE_FILE_FORMAT.key(), HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().toString());
- HoodieTableMetaClient metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ, properties);
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(MERGE_ON_READ, properties);
try (SparkRDDWriteClient client = getHoodieWriteClient(cfg)) {
@@ -319,6 +328,124 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
}
}
+ public static List<Arguments> testReattemptRollbackArguments() {
+ List<Arguments> arguments = new ArrayList<>();
+ for (boolean arg1 : new Boolean[] {true, false}) {
+ for (boolean arg2 : new Boolean[] {true, false}) {
+ arguments.add(Arguments.of(arg1, arg2));
+ }
+ }
+ return arguments;
+ }
+
+ @ParameterizedTest
+ @MethodSource("testReattemptRollbackArguments")
+ void testReattemptRollback(boolean rollbackUsingMarkers, boolean partitionedTable) throws Exception {
+ HoodieWriteConfig.Builder cfgBuilder =
+ getConfigBuilder(false, rollbackUsingMarkers, HoodieIndex.IndexType.SIMPLE);
+
+ addConfigsForPopulateMetaFields(cfgBuilder, true);
+ HoodieWriteConfig cfg = cfgBuilder.build();
+
+ Properties properties = CollectionUtils.copy(cfg.getProps());
+ properties.setProperty(HoodieTableConfig.BASE_FILE_FORMAT.key(), HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().toString());
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(MERGE_ON_READ, properties);
+
+ try (SparkRDDWriteClient client = getHoodieWriteClient(cfg)) {
+
+ HoodieTestDataGenerator dataGen = partitionedTable ? new HoodieTestDataGenerator()
+ : new HoodieTestDataGenerator(new String[] {NO_PARTITION_PATH});
+
+ // Test delta commit rollback
+ /*
+ * Write 1 (only inserts)
+ */
+ String newCommitTime = "000000001";
+ client.startCommitWithTime(newCommitTime);
+
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 200);
+ JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
+
+ JavaRDD<WriteStatus> writeStatusJavaRDD = client.upsert(writeRecords, newCommitTime);
+
+ List<WriteStatus> statuses = writeStatusJavaRDD.collect();
+ assertNoWriteErrors(statuses);
+
+ client.commit(newCommitTime, jsc().parallelize(statuses));
+
+ HoodieTable hoodieTable = HoodieSparkTable.create(cfg, context(), metaClient);
+
+ Option<HoodieInstant> deltaCommit = metaClient.getActiveTimeline().getDeltaCommitTimeline().firstInstant();
+ assertTrue(deltaCommit.isPresent());
+ assertEquals("000000001", deltaCommit.get().getTimestamp(), "Delta commit should be 000000001");
+
+ Option<HoodieInstant> commit = metaClient.getActiveTimeline().getCommitTimeline().firstInstant();
+ assertFalse(commit.isPresent());
+
+ FileStatus[] allFiles = listAllBaseFilesInPath(hoodieTable);
+ HoodieTableFileSystemView tableView = getHoodieTableFileSystemView(metaClient, metaClient.getCommitTimeline().filterCompletedInstants(), allFiles);
+ Stream<HoodieBaseFile> dataFilesToRead = tableView.getLatestBaseFiles();
+ assertFalse(dataFilesToRead.findAny().isPresent());
+
+ tableView = getHoodieTableFileSystemView(metaClient, hoodieTable.getCompletedCommitsTimeline(), allFiles);
+ dataFilesToRead = tableView.getLatestBaseFiles();
+ assertTrue(dataFilesToRead.findAny().isPresent(),
+ "should list the base files we wrote in the delta commit");
+
+ /*
+ * Write 2 (updates - testing failed delta commit)
+ */
+ final String commitTime1 = "000000002";
+ // WriteClient with custom config (disable small file handling)
+ try (SparkRDDWriteClient secondClient = getHoodieWriteClient(getHoodieWriteConfigWithSmallFileHandlingOff(true));) {
+ secondClient.startCommitWithTime(commitTime1);
+
+ List<HoodieRecord> copyOfRecords = new ArrayList<>(records);
+ copyOfRecords = dataGen.generateUpdates(commitTime1, copyOfRecords);
+
+ List<String> inputPaths = tableView.getLatestBaseFiles()
+ .map(baseFile -> new Path(baseFile.getPath()).getParent().toString())
+ .collect(Collectors.toList());
+ List<GenericRecord> recordsRead = HoodieMergeOnReadTestUtils.getRecordsUsingInputFormat(hadoopConf(), inputPaths,
+ basePath());
+ assertEquals(200, recordsRead.size());
+
+ statuses = secondClient.upsert(jsc().parallelize(copyOfRecords, 1), commitTime1).collect();
+ // Verify there are no errors
+ assertNoWriteErrors(statuses);
+
+ // simulate a failed rollback
+ String rollbackInstantTime = "000000003";
+ HoodieActiveTimeline activeTimeline = hoodieTable.getActiveTimeline().reload();
+ HoodieInstant failedDeltaCommitInstant = activeTimeline.getDeltaCommitTimeline().lastInstant().get();
+ assertEquals(commitTime1, failedDeltaCommitInstant.getTimestamp());
+ Option<HoodieRollbackPlan> rollbackPlan = hoodieTable.scheduleRollback(hoodieTable.getContext(), rollbackInstantTime,
+ failedDeltaCommitInstant, false, secondClient.getConfig().shouldRollbackUsingMarkers(), false);
+ assertTrue(rollbackPlan.isPresent());
+
+ MergeOnReadRollbackActionExecutor rollbackExecutor = new MergeOnReadRollbackActionExecutor<>(hoodieTable.getContext(),
+ secondClient.getConfig(), hoodieTable, rollbackInstantTime, failedDeltaCommitInstant, true, false);
+ List<HoodieRollbackStat> partialRollbackResult = rollbackExecutor.doRollbackAndGetStats(rollbackPlan.get());
+ // check that all partitions are included in this rollback
+ assertEquals(copyOfRecords.stream().map(HoodieRecord::getPartitionPath).distinct().count(), partialRollbackResult.size());
+
+ // do second rollback which should success
+ HoodieRollbackMetadata rollbackMetadata = hoodieTable.rollback(hoodieTable.getContext(), rollbackInstantTime, failedDeltaCommitInstant,
+ true, false);
+ HoodieRollbackStat rollbackStatInFirstTrial = partialRollbackResult.get(0);
+ HoodieRollbackPartitionMetadata rollbackPartitionMetadata = rollbackMetadata.getPartitionMetadata().get(rollbackStatInFirstTrial.getPartitionPath());
+
+ // check the log files generated in the first trial also appear in the second one.
+ Map<String, Long> commandLogBlockFiles = rollbackPartitionMetadata.getRollbackLogFiles();
+ for (FileStatus fileStatus : rollbackStatInFirstTrial.getCommandBlocksCount().keySet()) {
+ Long fileSize = commandLogBlockFiles.get(fileStatus.getPath().toString());
+ assertNotNull(fileSize);
+ assertEquals(fileStatus.getLen(), fileSize);
+ }
+ }
+ }
+ }
+
@Test
void testMultiRollbackWithDeltaAndCompactionCommit() throws Exception {
boolean populateMetaFields = true;
@@ -330,7 +457,7 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
Properties properties = getPropertiesForKeyGen(populateMetaFields);
properties.setProperty(HoodieTableConfig.BASE_FILE_FORMAT.key(), HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().toString());
- HoodieTableMetaClient metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ, properties);
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(MERGE_ON_READ, properties);
try (final SparkRDDWriteClient client = getHoodieWriteClient(cfg)) {
HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
@@ -512,7 +639,7 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
Properties properties = populateMetaFields ? new Properties() : getPropertiesForKeyGen();
properties.setProperty(HoodieTableConfig.BASE_FILE_FORMAT.key(), HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().toString());
- HoodieTableMetaClient metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ, properties);
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(MERGE_ON_READ, properties);
try (final SparkRDDWriteClient client = getHoodieWriteClient(cfg)) {
HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
@@ -531,7 +658,7 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
upsertRecords(client, "002", records, dataGen);
- client.savepoint("002","user1","comment1");
+ client.savepoint("002", "user1", "comment1");
upsertRecords(client, "003", records, dataGen);
upsertRecords(client, "004", records, dataGen);
@@ -586,7 +713,7 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
client.startCommitWithTime(commitTime);
List<HoodieRecord> copyOfRecords = new ArrayList<>(records);
copyOfRecords = dataGen.generateUpdates(commitTime, copyOfRecords);
- List<WriteStatus> statuses = client.upsert(jsc().parallelize(copyOfRecords, 1), commitTime).collect();
+ List<WriteStatus> statuses = client.upsert(jsc().parallelize(copyOfRecords, 1), commitTime).collect();
// Verify there are no errors
assertNoWriteErrors(statuses);
client.commit(commitTime, jsc().parallelize(statuses));
@@ -611,7 +738,7 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
properties.putAll(cfg.getProps());
properties.setProperty(HoodieTableConfig.BASE_FILE_FORMAT.key(), HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().toString());
- HoodieTableMetaClient metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ, properties);
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(MERGE_ON_READ, properties);
try (final SparkRDDWriteClient client = getHoodieWriteClient(cfg)) {
@@ -720,7 +847,7 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
void testInsertsGeneratedIntoLogFilesRollback(boolean rollbackUsingMarkers) throws Exception {
Properties properties = new Properties();
properties.setProperty(HoodieTableConfig.BASE_FILE_FORMAT.key(), HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().toString());
- HoodieTableMetaClient metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ, properties);
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(MERGE_ON_READ, properties);
HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
// insert 100 records
// Setting IndexType to be InMemory to simulate Global Index nature
@@ -813,7 +940,7 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
void testInsertsGeneratedIntoLogFilesRollbackAfterCompaction(boolean rollbackUsingMarkers) throws Exception {
Properties properties = new Properties();
properties.setProperty(HoodieTableConfig.BASE_FILE_FORMAT.key(), HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().toString());
- HoodieTableMetaClient metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ, properties);
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(MERGE_ON_READ, properties);
HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
// insert 100 records
// Setting IndexType to be InMemory to simulate Global Index nature
@@ -876,7 +1003,7 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
public void testLazyRollbackOfFailedCommit(boolean rollbackUsingMarkers) throws Exception {
Properties properties = new Properties();
properties.setProperty(HoodieTableConfig.BASE_FILE_FORMAT.key(), HoodieTableConfig.BASE_FILE_FORMAT.defaultValue().toString());
- HoodieTableMetaClient metaClient = getHoodieMetaClient(HoodieTableType.MERGE_ON_READ, properties);
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(MERGE_ON_READ, properties);
HoodieWriteConfig cfg = getWriteConfig(true, rollbackUsingMarkers);
HoodieWriteConfig autoCommitFalseCfg = getWriteConfig(false, rollbackUsingMarkers);
@@ -910,20 +1037,6 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
}
}
- private List<HoodieRecord> insertRecords(SparkRDDWriteClient client, HoodieTestDataGenerator dataGen, String commitTime) {
- /*
- * Write 1 (only inserts, written as base file)
- */
- client.startCommitWithTime(commitTime);
-
- List<HoodieRecord> records = dataGen.generateInserts(commitTime, 20);
- JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
-
- List<WriteStatus> statuses = client.upsert(writeRecords, commitTime).collect();
- assertNoWriteErrors(statuses);
- return records;
- }
-
private List<HoodieRecord> updateRecords(SparkRDDWriteClient client, HoodieTestDataGenerator dataGen, String commitTime,
List<HoodieRecord> records, HoodieTableMetaClient metaClient, HoodieWriteConfig cfg,
boolean assertLogFiles) throws IOException {
@@ -992,15 +1105,36 @@ public class TestHoodieSparkMergeOnReadTableRollback extends SparkClientFunction
return cfgBuilder.build();
}
- private SyncableFileSystemView getFileSystemViewWithUnCommittedSlices(HoodieTableMetaClient metaClient) {
- try {
- return new HoodieTableFileSystemView(metaClient,
- metaClient.getActiveTimeline(),
- HoodieTestTable.of(metaClient).listAllBaseAndLogFiles()
- );
- } catch (IOException ioe) {
- throw new HoodieIOException("Error getting file system view", ioe);
- }
+ /**
+ * Scenario: data table is updated, no changes to MDT
+ */
+ @Test
+ public void testRollbackWithFailurePreMDT() throws IOException {
+ testRollbackWithFailurePreMDT(MERGE_ON_READ);
+ }
+
+ /**
+ * Scenario: data table is updated, deltacommit is completed in MDT
+ */
+ @Test
+ public void testRollbackWithFailurePostMDT() throws IOException {
+ testRollbackWithFailurePostMDT(MERGE_ON_READ);
+ }
+
+ /**
+ * Scenario: data table is updated, deltacommit is completed in MDT then during rollback,
+ * data table is updated, no changes to MDT
+ */
+ @Test
+ public void testRollbackWithFailurePostMDTRollbackFailsPreMDT() throws IOException {
+ testRollbackWithFailurePostMDT(MERGE_ON_READ, true);
}
+ /**
+ * Scenario: data table is updated, deltacommit of interest is inflight in MDT
+ */
+ @Test
+ public void testRollbackWithFailureInMDT() throws Exception {
+ testRollbackWithFailureinMDT(MERGE_ON_READ);
+ }
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkRollback.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkRollback.java
new file mode 100644
index 00000000000..174ec63a23b
--- /dev/null
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkRollback.java
@@ -0,0 +1,287 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.table.functional;
+
+import org.apache.hudi.avro.model.HoodieRollbackMetadata;
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.client.functional.TestHoodieBackedMetadata;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieDeltaWriteStat;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.HoodieWriteStat;
+import org.apache.hudi.common.model.IOType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.table.HoodieSparkTable;
+import org.apache.hudi.table.marker.WriteMarkers;
+import org.apache.hudi.table.marker.WriteMarkersFactory;
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
+
+import org.apache.spark.api.java.JavaRDD;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.List;
+
+import static org.apache.hudi.common.model.HoodieTableType.COPY_ON_WRITE;
+import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
+import static org.apache.hudi.testutils.Assertions.assertNoWriteErrors;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TestHoodieSparkRollback extends SparkClientFunctionalTestHarness {
+
+ private String basePath;
+
+ private void initBasePath() {
+ basePath = basePath().substring(7);
+ }
+
+ private SparkRDDWriteClient getHoodieWriteClient(Boolean autoCommitEnabled) throws IOException {
+ return getHoodieWriteClient(getConfigToTestMDTRollbacks(autoCommitEnabled));
+ }
+
+ protected List<HoodieRecord> insertRecords(SparkRDDWriteClient client, HoodieTestDataGenerator dataGen, String commitTime) {
+ /*
+ * Write 1 (only inserts, written as base file)
+ */
+ client.startCommitWithTime(commitTime);
+
+ List<HoodieRecord> records = dataGen.generateInserts(commitTime, 20);
+ JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
+
+ List<WriteStatus> statuses = client.upsert(writeRecords, commitTime).collect();
+ assertNoWriteErrors(statuses);
+ return records;
+ }
+
+ protected List<WriteStatus> updateRecords(SparkRDDWriteClient client, HoodieTestDataGenerator dataGen, String commitTime,
+ List<HoodieRecord> records) throws IOException {
+ client.startCommitWithTime(commitTime);
+
+ records = dataGen.generateUpdates(commitTime, records);
+ JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
+ List<WriteStatus> statuses = client.upsert(writeRecords, commitTime).collect();
+ assertNoWriteErrors(statuses);
+ return statuses;
+ }
+
+ protected HoodieWriteConfig getConfigToTestMDTRollbacks(Boolean autoCommit) {
+ return getConfigToTestMDTRollbacks(autoCommit, true);
+ }
+
+ protected HoodieWriteConfig getConfigToTestMDTRollbacks(Boolean autoCommit, Boolean mdtEnable) {
+ return HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .withProperties(getPropertiesForKeyGen(true))
+ .withSchema(TRIP_EXAMPLE_SCHEMA)
+ .withParallelism(2, 2)
+ .withDeleteParallelism(2)
+ .withAutoCommit(autoCommit)
+ .withEmbeddedTimelineServerEnabled(false).forTable("test-trip-table")
+ .withRollbackUsingMarkers(true)
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(mdtEnable).build())
+ .build();
+ }
+
+ /**
+ * Scenario: data table is updated, no changes to MDT
+ */
+ protected void testRollbackWithFailurePreMDT(HoodieTableType tableType) throws IOException {
+ initBasePath();
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(tableType);
+ SparkRDDWriteClient client = getHoodieWriteClient(true);
+ HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
+
+ //normal insert
+ List<HoodieRecord> records = insertRecords(client, dataGen, "001");
+ //update but don't commit
+ client = getHoodieWriteClient(false);
+ updateRecords(client, dataGen, "002", records);
+ //New update will trigger rollback and we will commit this time
+ client = getHoodieWriteClient(true);
+ updateRecords(client, dataGen, "003", records);
+ //validate that metadata table file listing matches reality
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ TestHoodieBackedMetadata.validateMetadata(getConfigToTestMDTRollbacks(true), Option.empty(), fs(), basePath, metaClient,
+ hadoopConf(), new HoodieSparkEngineContext(jsc()), TestHoodieBackedMetadata.metadata(client));
+ }
+
+ /**
+ * Scenario: data table is updated, deltacommit is completed in MDT
+ */
+ protected void testRollbackWithFailurePostMDT(HoodieTableType tableType) throws IOException {
+ testRollbackWithFailurePostMDT(tableType, false);
+ }
+
+ protected void testRollbackWithFailurePostMDT(HoodieTableType tableType, Boolean failRollback) throws IOException {
+ initBasePath();
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(tableType);
+ HoodieWriteConfig cfg = getConfigToTestMDTRollbacks(true);
+ SparkRDDWriteClient client = getHoodieWriteClient(cfg);
+ HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
+
+ //normal insert
+ List<HoodieRecord> records = insertRecords(client, dataGen, "001");
+ //New update and commit so that the MDT has the update
+ List<WriteStatus> statuses = updateRecords(client, dataGen, "002", records);
+
+ //delete commit from timeline
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ String filename = metaClient.getActiveTimeline().lastInstant().get().getFileName();
+ File commit = new File(metaClient.getBasePathV2().toString().substring(5) + "/.hoodie/" + filename);
+ assertTrue(commit.delete());
+ metaClient.reloadActiveTimeline();
+
+ //Add back the marker files to mimic that we haven't committed yet
+ statuses.forEach(s -> {
+ try {
+ recreateMarkerFile(cfg, "002", s);
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ if (failRollback) {
+ copyOut(tableType, "002");
+ //disable MDT so we don't copy it
+ client = getHoodieWriteClient(getConfigToTestMDTRollbacks(true, false));
+ assertTrue(client.rollback("002", "003"));
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieInstant lastInstant = metaClient.getActiveTimeline().lastInstant().get();
+ assertEquals(HoodieTimeline.ROLLBACK_ACTION, lastInstant.getAction());
+ HoodieRollbackMetadata rollbackMetadata = TimelineMetadataUtils.deserializeHoodieRollbackMetadata(metaClient.getActiveTimeline().getInstantDetails(lastInstant).get());
+ copyIn(tableType, "002");
+ rollbackMetadata.getPartitionMetadata().forEach((partition, metadata) -> metadata.getRollbackLogFiles().forEach((n, k) -> recreateMarkerFile(cfg, "003", partition, n)));
+ rollbackMetadata.getPartitionMetadata().forEach((partition, metadata) -> metadata.getLogFilesFromFailedCommit().forEach((n, k) -> recreateMarkerFile(cfg, "002", partition, n)));
+ commit = new File(metaClient.getBasePathV2().toString().substring(5) + "/.hoodie/" + lastInstant.getFileName());
+ assertTrue(commit.delete());
+ metaClient.reloadActiveTimeline();
+ }
+
+ //now we are at a state that we would be at if a write failed after writing to MDT but before commit is finished
+
+ //New update will trigger rollback and we will commit this time
+ client = getHoodieWriteClient(getConfigToTestMDTRollbacks(true, true));
+ updateRecords(client, dataGen, "004", records);
+ //validate that metadata table file listing matches reality
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ TestHoodieBackedMetadata.validateMetadata(cfg, Option.empty(), fs(), basePath, metaClient, hadoopConf(), new HoodieSparkEngineContext(jsc()), TestHoodieBackedMetadata.metadata(client));
+ }
+
+ private void copyOut(HoodieTableType tableType, String commitTime) throws IOException {
+ File tmpDir = new File(basePath, ".tmpdir");
+ assertTrue(tmpDir.mkdir());
+ String commitAction = (tableType.equals(COPY_ON_WRITE) ? ".commit" : ".deltacommit");
+ String metaDir = basePath + ".hoodie/";
+ String inflight = commitTime + (tableType.equals(COPY_ON_WRITE) ? "" : commitAction) + ".inflight";
+ Files.copy(new File(metaDir + inflight).toPath(), tmpDir.toPath().resolve(inflight), StandardCopyOption.REPLACE_EXISTING);
+ String requested = commitTime + commitAction + ".requested";
+ Files.copy(new File(metaDir + requested).toPath(), tmpDir.toPath().resolve(requested), StandardCopyOption.REPLACE_EXISTING);
+ }
+
+ private void copyIn(HoodieTableType tableType, String commitTime) throws IOException {
+ Path tmpDir = new File(basePath, ".tmpdir").toPath();
+ String commitAction = (tableType.equals(COPY_ON_WRITE) ? ".commit" : ".deltacommit");
+ String metaDir = basePath + ".hoodie/";
+ String inflight = commitTime + (tableType.equals(COPY_ON_WRITE) ? "" : commitAction) + ".inflight";
+ Files.copy(tmpDir.resolve(inflight), new File(metaDir + inflight).toPath(), StandardCopyOption.REPLACE_EXISTING);
+ String requested = commitTime + commitAction + ".requested";
+ Files.copy(tmpDir.resolve(requested), new File(metaDir + requested).toPath(), StandardCopyOption.REPLACE_EXISTING);
+ }
+
+ /**
+ * Scenario: data table is updated, deltacommit of interest is inflight in MDT
+ */
+ protected void testRollbackWithFailureinMDT(HoodieTableType tableType) throws Exception {
+ initBasePath();
+ HoodieWriteConfig cfg = getConfigToTestMDTRollbacks(true);
+ HoodieTableMetaClient metaClient = getHoodieMetaClient(tableType);
+ SparkRDDWriteClient client = getHoodieWriteClient(cfg);
+ HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator();
+
+ //normal insert
+ List<HoodieRecord> records = insertRecords(client, dataGen, "001");
+ //New update and commit
+ List<WriteStatus> statuses = updateRecords(client, dataGen, "002", records);
+
+ //delete commit from timeline
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ String filename = metaClient.getActiveTimeline().lastInstant().get().getFileName();
+ File deltacommit = new File(metaClient.getBasePathV2().toString().substring(5) + "/.hoodie/" + filename);
+ assertTrue(deltacommit.delete());
+ metaClient.reloadActiveTimeline();
+
+ //Add back the marker files to mimic that we haven't committed yet
+ statuses.forEach(s -> {
+ try {
+ recreateMarkerFile(cfg, "002", s);
+ } catch (IOException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ //Make the MDT appear to fail mid write by deleting the commit in the MDT timline. The MDT does not use markers so we do not need to recreate them
+ String metadataBasePath = basePath + "/.hoodie/metadata";
+ HoodieTableMetaClient metadataMetaClient = HoodieTableMetaClient.builder().setConf(hadoopConf()).setBasePath(metadataBasePath).build();
+ HoodieInstant latestCommitInstant = metadataMetaClient.getActiveTimeline().lastInstant().get();
+ File metadatadeltacommit = new File(metadataBasePath + "/.hoodie/" + latestCommitInstant.getFileName());
+ assertTrue(metadatadeltacommit.delete());
+
+ //New update will trigger rollback and we will commit this time
+ updateRecords(client, dataGen, "003", records);
+ //validate that metadata table file listing matches reality
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ TestHoodieBackedMetadata.validateMetadata(cfg, Option.empty(), fs(), basePath, metaClient,
+ hadoopConf(), new HoodieSparkEngineContext(jsc()), TestHoodieBackedMetadata.metadata(client));
+ }
+
+ /**
+ * We are simulating scenarios where commits fail inflight. To mimic this, we need to recreate the marker files for the files that are
+ * written in the "failed" commit
+ * */
+ protected void recreateMarkerFile(HoodieWriteConfig cfg, String commitTime, WriteStatus writeStatus) throws IOException, InterruptedException {
+ HoodieWriteStat writeStat = writeStatus.getStat();
+ final WriteMarkers writeMarkers = WriteMarkersFactory.get(cfg.getMarkersType(),
+ HoodieSparkTable.create(cfg, context()), commitTime);
+ if (writeStat instanceof HoodieDeltaWriteStat) {
+ ((HoodieDeltaWriteStat) writeStat).getLogFiles().forEach(lf -> writeMarkers.create(writeStat.getPartitionPath(), lf, IOType.APPEND));
+ } else {
+ writeMarkers.create(writeStat.getPartitionPath(), writeStat.getPath().replace(writeStat.getPartitionPath() + "/",""), IOType.MERGE);
+ }
+ }
+
+ protected void recreateMarkerFile(HoodieWriteConfig cfg, String commitTime, String partitionPath, String path) {
+ final WriteMarkers writeMarkers = WriteMarkersFactory.get(cfg.getMarkersType(),
+ HoodieSparkTable.create(cfg, context()), commitTime);
+ writeMarkers.create(partitionPath, new File(path).getName(), IOType.APPEND);
+ }
+}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestMarkerBasedRollbackStrategy.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestMarkerBasedRollbackStrategy.java
index d806347b682..f1c78dc877a 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestMarkerBasedRollbackStrategy.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestMarkerBasedRollbackStrategy.java
@@ -47,6 +47,7 @@ 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.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
import java.io.IOException;
@@ -87,11 +88,14 @@ public class TestMarkerBasedRollbackStrategy extends HoodieClientTestBase {
@Test
public void testMarkerBasedRollbackAppend() throws Exception {
+ tearDown();
+ tableType = HoodieTableType.MERGE_ON_READ;
+ setUp();
HoodieTestTable testTable = HoodieTestTable.of(metaClient);
String f0 = testTable.addRequestedCommit("000")
.getFileIdsWithBaseFilesInPartitions("partA").get("partA");
testTable.forCommit("001")
- .withMarkerFile("partA", f0, IOType.APPEND);
+ .withLogMarkerFile("partA", f0, IOType.APPEND);
HoodieTable hoodieTable = HoodieSparkTable.create(getConfig(), context, metaClient);
List<HoodieRollbackRequest> rollbackRequests = new MarkerBasedRollbackStrategy(hoodieTable, context, getConfig(),
@@ -99,6 +103,29 @@ public class TestMarkerBasedRollbackStrategy extends HoodieClientTestBase {
assertEquals(1, rollbackRequests.size());
}
+ @ParameterizedTest
+ @EnumSource(names = {"APPEND"})
+ public void testMarkerBasedRollbackAppendWithLogFileMarkers(IOType testIOType) throws Exception {
+ tearDown();
+ tableType = HoodieTableType.MERGE_ON_READ;
+ setUp();
+ HoodieTestTable testTable = HoodieTestTable.of(metaClient);
+ String f0 = testTable.addRequestedCommit("000")
+ .getFileIdWithLogFile("partA");
+ testTable.forCommit("001")
+ .withLogMarkerFile("partA", f0, testIOType);
+
+ HoodieTable hoodieTable = HoodieSparkTable.create(getConfig(), context, metaClient);
+ List<HoodieRollbackRequest> rollbackRequests = new MarkerBasedRollbackStrategy(hoodieTable, context, getConfig(), "002")
+ .getRollbackRequests(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, "001"));
+ assertEquals(1, rollbackRequests.size());
+ HoodieRollbackRequest rollbackRequest = rollbackRequests.get(0);
+ assertEquals("partA", rollbackRequest.getPartitionPath());
+ assertEquals(f0, rollbackRequest.getFileId());
+ assertEquals(testIOType.equals(IOType.CREATE) ? 1 : 0, rollbackRequest.getFilesToBeDeleted().size());
+ assertEquals(1, rollbackRequest.getLogBlocksToBeDeleted().size());
+ }
+
@Test
public void testCopyOnWriteRollbackWithTestTable() throws Exception {
// given: wrote some base files and corresponding markers
@@ -115,11 +142,11 @@ public class TestMarkerBasedRollbackStrategy extends HoodieClientTestBase {
.withMarkerFile("partA", f2, IOType.CREATE);
// when
- HoodieTable hoodieTable = HoodieSparkTable.create(getConfig(), context, metaClient);
+ HoodieTable hoodieTable = HoodieSparkTable.create(getConfigBuilder().withEmbeddedTimelineServerEnabled(false).build(), context, metaClient);
List<HoodieRollbackRequest> rollbackRequests = new MarkerBasedRollbackStrategy(hoodieTable, context, getConfig(),
"002").getRollbackRequests(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.COMMIT_ACTION, "001"));
- List<HoodieRollbackStat> stats = new BaseRollbackHelper(hoodieTable.getMetaClient(), getConfig()).performRollback(context,
+ List<HoodieRollbackStat> stats = new BaseRollbackHelper(hoodieTable, getConfig()).performRollback(context, "002",
new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.COMMIT_ACTION, "001"),
rollbackRequests);
@@ -139,6 +166,7 @@ public class TestMarkerBasedRollbackStrategy extends HoodieClientTestBase {
@MethodSource("configParams")
public void testCopyOnWriteRollback(boolean useFileListingMetadata) throws Exception {
HoodieWriteConfig writeConfig = getConfigBuilder().withRollbackUsingMarkers(true).withAutoCommit(false)
+ .withEmbeddedTimelineServerEnabled(false)
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(useFileListingMetadata).build())
.withPath(basePath).build();
@@ -164,7 +192,8 @@ public class TestMarkerBasedRollbackStrategy extends HoodieClientTestBase {
tableType = HoodieTableType.MERGE_ON_READ;
setUp();
- HoodieWriteConfig writeConfig = getConfigBuilder().withRollbackUsingMarkers(true).withAutoCommit(false)
+ HoodieWriteConfig writeConfig = getConfigBuilder().withRollbackUsingMarkers(true)
+ .withEmbeddedTimelineServerEnabled(false).withAutoCommit(false)
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(useFileListingMetadata).build())
.withPath(basePath).build();
@@ -193,6 +222,7 @@ public class TestMarkerBasedRollbackStrategy extends HoodieClientTestBase {
setUp();
HoodieWriteConfig writeConfig = getConfigBuilder().withRollbackUsingMarkers(true).withAutoCommit(false)
+ .withEmbeddedTimelineServerEnabled(false)
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(useFileListingMetadata).build())
.withPath(basePath).build();
@@ -222,12 +252,13 @@ public class TestMarkerBasedRollbackStrategy extends HoodieClientTestBase {
writeStatuses.collect();
- HoodieTable hoodieTable = HoodieSparkTable.create(getConfig(), context, metaClient);
- List<HoodieRollbackRequest> rollbackRequests = new MarkerBasedRollbackStrategy(hoodieTable, context, getConfig(),
+ HoodieTable hoodieTable = HoodieSparkTable.create(getConfigBuilder().withEmbeddedTimelineServerEnabled(false).build(), context, metaClient);
+ List<HoodieRollbackRequest> rollbackRequests = new MarkerBasedRollbackStrategy(hoodieTable, context,
+ getConfigBuilder().withEmbeddedTimelineServerEnabled(false).build(),
"002").getRollbackRequests(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, "001"));
// rollback 1st commit and ensure stats reflect the info.
- return new BaseRollbackHelper(hoodieTable.getMetaClient(), getConfig()).performRollback(context,
+ return new BaseRollbackHelper(hoodieTable, getConfig()).performRollback(context, "002",
new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, "001"),
rollbackRequests);
}
@@ -247,12 +278,13 @@ public class TestMarkerBasedRollbackStrategy extends HoodieClientTestBase {
writeStatuses = writeClient.upsert(jsc.parallelize(records, 1), newCommitTime);
writeStatuses.collect();
- HoodieTable hoodieTable = HoodieSparkTable.create(getConfig(), context, metaClient);
- List<HoodieRollbackRequest> rollbackRequests = new MarkerBasedRollbackStrategy(hoodieTable, context, getConfig(),
+ HoodieTable hoodieTable = HoodieSparkTable.create(getConfigBuilder().withEmbeddedTimelineServerEnabled(false).build(), context, metaClient);
+ List<HoodieRollbackRequest> rollbackRequests = new MarkerBasedRollbackStrategy(hoodieTable, context, getConfigBuilder()
+ .withEmbeddedTimelineServerEnabled(false).build(),
"003").getRollbackRequests(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, "002"));
// rollback 2nd commit and ensure stats reflect the info.
- return new BaseRollbackHelper(hoodieTable.getMetaClient(), getConfig()).performRollback(context,
+ return new BaseRollbackHelper(hoodieTable, getConfig()).performRollback(context, "003",
new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.DELTA_COMMIT_ACTION, "002"),
rollbackRequests);
}
@@ -263,7 +295,7 @@ public class TestMarkerBasedRollbackStrategy extends HoodieClientTestBase {
String f0 = testTable.addRequestedCommit("000")
.getFileIdsWithBaseFilesInPartitions("partA").get("partA");
testTable.forCommit("001")
- .withMarkerFile("partA", f0, IOType.APPEND);
+ .withLogMarkerFile("partA", f0, IOType.APPEND);
HoodieTable hoodieTable = HoodieSparkTable.create(getConfig(), context, metaClient);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestWriteMarkersBase.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestWriteMarkersBase.java
index 6ba783c749f..c0f057ffb86 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestWriteMarkersBase.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestWriteMarkersBase.java
@@ -121,6 +121,27 @@ public abstract class TestWriteMarkersBase extends HoodieCommonTestHarness {
);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testGetAppendedLogPaths(boolean isTablePartitioned) throws IOException {
+ // add marker files
+ createSomeMarkers(isTablePartitioned);
+ // add invalid file
+ createInvalidFile(isTablePartitioned ? "2020/06/01" : "", "invalid_file3");
+ long fileSize = FileSystemTestUtils.listRecursive(fs, markerFolderPath).stream()
+ .filter(fileStatus -> !fileStatus.getPath().getName().contains(MarkerUtils.MARKER_TYPE_FILENAME))
+ .count();
+ assertEquals(fileSize, 4);
+
+ List<String> expectedPaths = isTablePartitioned
+ ? CollectionUtils.createImmutableList("2020/06/02/file2")
+ : CollectionUtils.createImmutableList("file2");
+ // then
+ assertIterableEquals(expectedPaths,
+ writeMarkers.getAppendedLogPaths(context, 2).stream().sorted().collect(Collectors.toList())
+ );
+ }
+
@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testAllMarkerPaths(boolean isTablePartitioned) throws IOException {
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
index 111b2141e28..81e498758a9 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/upgrade/TestUpgradeDowngrade.java
@@ -22,11 +22,13 @@ import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.FileSlice;
import org.apache.hudi.common.model.HoodieFileGroup;
import org.apache.hudi.common.model.HoodieLogFile;
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.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableVersion;
@@ -200,6 +202,7 @@ public class TestUpgradeDowngrade extends HoodieClientTestBase {
Pair<List<HoodieRecord>, List<HoodieRecord>> inputRecords = twoUpsertCommitDataWithTwoPartitions(firstPartitionCommit2FileSlices, secondPartitionCommit2FileSlices, cfg, client, false);
HoodieTable table = this.getHoodieTable(metaClient, cfg);
+ prepForUpgradeFromZeroToOne(table);
HoodieInstant commitInstant = table.getPendingCommitTimeline().lastInstant().get();
// delete one of the marker files in 2nd commit if need be.
@@ -844,6 +847,47 @@ public class TestUpgradeDowngrade extends HoodieClientTestBase {
return Pair.of(records, records2);
}
+ /**
+ * Since how markers are generated for log file changed in Version Six, we regenerate markers in the way version zero do.
+ *
+ * @param table instance of {@link HoodieTable}
+ */
+ private void prepForUpgradeFromZeroToOne(HoodieTable table) throws IOException {
+ List<HoodieInstant> instantsToBeParsed =
+ metaClient.getActiveTimeline()
+ .getCommitsTimeline()
+ .getInstantsAsStream()
+ .collect(Collectors.toList());
+ for (HoodieInstant instant : instantsToBeParsed) {
+ WriteMarkers writeMarkers =
+ WriteMarkersFactory.get(table.getConfig().getMarkersType(), table, instant.getTimestamp());
+ Set<String> oldMarkers = writeMarkers.allMarkerFilePaths();
+ boolean hasAppendMarker = oldMarkers.stream().anyMatch(marker -> marker.contains(IOType.APPEND.name()));
+ if (hasAppendMarker) {
+ // delete all markers and regenerate
+ writeMarkers.deleteMarkerDir(table.getContext(), 2);
+ for (String oldMarker : oldMarkers) {
+ String typeStr = oldMarker.substring(oldMarker.lastIndexOf(".") + 1);
+ IOType type = IOType.valueOf(typeStr);
+ String partitionFilePath = WriteMarkers.stripMarkerSuffix(oldMarker);
+ Path fullFilePath = new Path(basePath, partitionFilePath);
+ String partitionPath = FSUtils.getRelativePartitionPath(new Path(basePath), fullFilePath.getParent());
+ if (FSUtils.isBaseFile(fullFilePath)) {
+ writeMarkers.create(partitionPath, fullFilePath.getName(), type);
+ } else {
+ String fileId = FSUtils.getFileIdFromFilePath(fullFilePath);
+ String baseInstant = FSUtils.getBaseCommitTimeFromLogPath(fullFilePath);
+ String writeToken = FSUtils.getWriteTokenFromLogPath(fullFilePath);
+ writeMarkers.create(partitionPath,
+ FSUtils.makeBaseFileName(baseInstant, writeToken, fileId, table.getBaseFileFormat().getFileExtension()), type);
+ }
+ }
+ writeMarkers.allMarkerFilePaths()
+ .forEach(markerPath -> assertFalse(markerPath.contains(HoodieLogFile.DELTA_EXTENSION)));
+ }
+ }
+ }
+
private void prepForDowngradeFromVersion(HoodieTableVersion fromVersion) throws IOException {
metaClient.getTableConfig().setTableVersion(fromVersion);
Path propertyFile = new Path(metaClient.getMetaPath() + "/" + HoodieTableConfig.HOODIE_PROPERTIES_FILE);
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieListPairData.java b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieListPairData.java
index 39ce1411575..af73a3cbad6 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieListPairData.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodieListPairData.java
@@ -26,6 +26,8 @@ import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.common.util.collection.MappingIterator;
import org.apache.hudi.common.util.collection.Pair;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
@@ -191,6 +193,31 @@ public class HoodieListPairData<K, V> extends HoodieBaseListData<Pair<K, V>> imp
return new HoodieListPairData<>(leftOuterJoined, lazy);
}
+ @Override
+ public <W> HoodiePairData<K, Pair<V, W>> join(HoodiePairData<K, W> other) {
+ ValidationUtils.checkArgument(other instanceof HoodieListPairData);
+
+ // Transform right-side container to a multi-map of [[K]] to [[List<W>]] values
+ HashMap<K, List<W>> rightStreamMap = ((HoodieListPairData<K, W>) other).asStream().collect(
+ Collectors.groupingBy(
+ Pair::getKey,
+ HashMap::new,
+ Collectors.mapping(Pair::getValue, Collectors.toList())));
+
+ List<Pair<K, Pair<V, W>>> joinResult = new ArrayList<>();
+ asStream().forEach(pair -> {
+ K key = pair.getKey();
+ V leftValue = pair.getValue();
+ List<W> rightValues = rightStreamMap.getOrDefault(key, Collections.emptyList());
+
+ for (W rightValue : rightValues) {
+ joinResult.add(Pair.of(key, Pair.of(leftValue, rightValue)));
+ }
+ });
+
+ return new HoodieListPairData<>(joinResult, lazy);
+ }
+
@Override
public long count() {
return super.count();
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodiePairData.java b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodiePairData.java
index 1d3622786fd..de010f80445 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/data/HoodiePairData.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/data/HoodiePairData.java
@@ -123,6 +123,18 @@ public interface HoodiePairData<K, V> extends Serializable {
*/
<W> HoodiePairData<K, Pair<V, Option<W>>> leftOuterJoin(HoodiePairData<K, W> other);
+ /**
+ * Performs an inner join of this dataset against {@code other}.
+ *
+ * For each element (k, v) in this, the resulting {@link HoodiePairData} will contain all
+ * pairs {@code (k, (v, Some(w)))} for every {@code w} in the {@code other},
+ *
+ * @param other the other {@link HoodiePairData}
+ * @param <W> value type of the other {@link HoodiePairData}
+ * @return containing the result of the left outer join
+ */
+ <W> HoodiePairData<K, Pair<V, W>> join(HoodiePairData<K, W> other);
+
/**
* Collects results of the underlying collection into a {@link List<Pair<K, V>>}
*
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 1d72d706371..a090eb8544f 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
@@ -286,6 +286,63 @@ public class FSUtils {
}
}
+ /**
+ * Get all the files in the given partition path.
+ *
+ * @param fileSystem File System
+ * @param partitionPathIncludeBasePath The full partition path including the base path
+ * @param filesNamesUnderThisPartition The names of the files under this partition for which file status is needed
+ * @param ignoreMissingFiles If true, missing files will be ignored and empty Option will be added to the result list
+ * @return List of file statuses for the files under this partition
+ */
+ public static List<Option<FileStatus>> getFileStatusesUnderPartition(FileSystem fileSystem,
+ Path partitionPathIncludeBasePath,
+ Set<String> filesNamesUnderThisPartition,
+ boolean ignoreMissingFiles) {
+ String fileSystemType = fileSystem.getScheme();
+ boolean useListStatus = StorageSchemes.isListStatusFriendly(fileSystemType);
+ List<Option<FileStatus>> result = new ArrayList<>(filesNamesUnderThisPartition.size());
+ try {
+ if (useListStatus) {
+ FileStatus[] fileStatuses = fileSystem.listStatus(partitionPathIncludeBasePath,
+ path -> filesNamesUnderThisPartition.contains(path.getName()));
+ Map<String, FileStatus> filenameToFileStatusMap = Arrays.stream(fileStatuses)
+ .collect(Collectors.toMap(
+ fileStatus -> fileStatus.getPath().getName(),
+ fileStatus -> fileStatus
+ ));
+
+ for (String fileName : filesNamesUnderThisPartition) {
+ if (filenameToFileStatusMap.containsKey(fileName)) {
+ result.add(Option.of(filenameToFileStatusMap.get(fileName)));
+ } else {
+ if (!ignoreMissingFiles) {
+ throw new FileNotFoundException("File not found: " + new Path(partitionPathIncludeBasePath.toString(), fileName));
+ }
+ result.add(Option.empty());
+ }
+ }
+ } else {
+ for (String fileName : filesNamesUnderThisPartition) {
+ Path fullPath = new Path(partitionPathIncludeBasePath.toString(), fileName);
+ try {
+ FileStatus fileStatus = fileSystem.getFileStatus(fullPath);
+ result.add(Option.of(fileStatus));
+ } catch (FileNotFoundException fileNotFoundException) {
+ if (ignoreMissingFiles) {
+ result.add(Option.empty());
+ } else {
+ throw new FileNotFoundException("File not found: " + fullPath.toString());
+ }
+ }
+ }
+ }
+ } catch (IOException e) {
+ throw new HoodieIOException("List files under " + partitionPathIncludeBasePath + " failed", e);
+ }
+ return result;
+ }
+
public static String getFileExtension(String fullName) {
Objects.requireNonNull(fullName);
String fileName = new File(fullName).getName();
@@ -496,6 +553,7 @@ public class FSUtils {
public static Stream<HoodieLogFile> getAllLogFiles(FileSystem fs, Path partitionPath, final String fileId,
final String logFileExtension, final String baseCommitTime) throws IOException {
try {
+ // TODO: Use a better filter to avoid listing all files i.e. use baseCommitTime in the filter too.
PathFilter pathFilter = path -> path.getName().startsWith("." + fileId) && path.getName().contains(logFileExtension);
return Arrays.stream(fs.listStatus(partitionPath, pathFilter))
.map(HoodieLogFile::new)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileWriteCallback.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileWriteCallback.java
new file mode 100644
index 00000000000..652c013cc3e
--- /dev/null
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileWriteCallback.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.table.log;
+
+import org.apache.hudi.common.model.HoodieLogFile;
+
+/**
+ * HoodieLogFileWriteCallback is trigger when specific log file operation happen
+ */
+public interface HoodieLogFileWriteCallback {
+ default boolean preLogFileOpen(HoodieLogFile logFileToAppend) {
+ return true;
+ }
+
+ default boolean preLogFileCreate(HoodieLogFile logFileToCreate) {
+ return true;
+ }
+
+ default boolean preLogFileClose(HoodieLogFile logFileToClose) {
+ return true;
+ }
+
+ default boolean postLogFileClose(HoodieLogFile logFileToClose) {
+ return true;
+ }
+}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormat.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormat.java
index d77be9a281b..5e7d0806fae 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormat.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormat.java
@@ -145,6 +145,8 @@ public interface HoodieLogFormat {
private String suffix;
// Rollover Log file write token
private String rolloverLogWriteToken;
+ // A call back triggered with log file operation
+ private HoodieLogFileWriteCallback logFileWriteCallback;
public WriterBuilder withBufferSize(int bufferSize) {
this.bufferSize = bufferSize;
@@ -201,6 +203,11 @@ public interface HoodieLogFormat {
return this;
}
+ public WriterBuilder withLogWriteCallback(HoodieLogFileWriteCallback logFileWriteCallback) {
+ this.logFileWriteCallback = logFileWriteCallback;
+ return this;
+ }
+
public WriterBuilder withFileSize(long fileLen) {
this.fileLen = fileLen;
return this;
@@ -233,6 +240,11 @@ public interface HoodieLogFormat {
rolloverLogWriteToken = UNKNOWN_WRITE_TOKEN;
}
+ if (logFileWriteCallback == null) {
+ // use a callback do nothing here as default callback.
+ logFileWriteCallback = new HoodieLogFileWriteCallback() {};
+ }
+
if (logVersion == null) {
LOG.info("Computing the next log version for " + logFileId + " in " + parentPath);
Option<Pair<Integer, String>> versionAndWriteToken =
@@ -279,7 +291,8 @@ public interface HoodieLogFormat {
if (sizeThreshold == null) {
sizeThreshold = DEFAULT_SIZE_THRESHOLD;
}
- return new HoodieLogFormatWriter(fs, logFile, bufferSize, replication, sizeThreshold, rolloverLogWriteToken);
+ return new HoodieLogFormatWriter(fs, logFile, bufferSize, replication, sizeThreshold,
+ rolloverLogWriteToken, logFileWriteCallback);
}
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatWriter.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatWriter.java
index fd4f24f89d8..0b16d2ee2a6 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatWriter.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFormatWriter.java
@@ -55,18 +55,21 @@ public class HoodieLogFormatWriter implements HoodieLogFormat.Writer {
private final Integer bufferSize;
private final Short replication;
private final String rolloverLogWriteToken;
+ final HoodieLogFileWriteCallback logFileWriteCallback;
private boolean closed = false;
private transient Thread shutdownThread = null;
private static final String APPEND_UNAVAILABLE_EXCEPTION_MESSAGE = "not sufficiently replicated yet";
- HoodieLogFormatWriter(FileSystem fs, HoodieLogFile logFile, Integer bufferSize, Short replication, Long sizeThreshold, String rolloverLogWriteToken) {
+ HoodieLogFormatWriter(FileSystem fs, HoodieLogFile logFile, Integer bufferSize, Short replication, Long sizeThreshold,
+ String rolloverLogWriteToken, HoodieLogFileWriteCallback logFileWriteCallback) {
this.fs = fs;
this.logFile = logFile;
this.sizeThreshold = sizeThreshold;
this.bufferSize = bufferSize;
this.replication = replication;
this.rolloverLogWriteToken = rolloverLogWriteToken;
+ this.logFileWriteCallback = logFileWriteCallback;
addShutDownHook();
}
@@ -94,7 +97,9 @@ public class HoodieLogFormatWriter implements HoodieLogFormat.Writer {
Path path = logFile.getPath();
if (fs.exists(path)) {
boolean isAppendSupported = StorageSchemes.isAppendSupported(fs.getScheme());
- if (isAppendSupported) {
+ // here we use marker file to fence concurrent append to the same file. So it is safe to use speculation in spark now.
+ boolean canAppend = isAppendSupported ? logFileWriteCallback.preLogFileOpen(logFile) : false;
+ if (canAppend) {
LOG.info(logFile + " exists. Appending to existing file");
try {
// open the path for append and record the offset
@@ -116,10 +121,11 @@ public class HoodieLogFormatWriter implements HoodieLogFormat.Writer {
}
}
}
- if (!isAppendSupported) {
+ if (!isAppendSupported || !canAppend) {
rollOver();
createNewFile();
- LOG.info("Append not supported.. Rolling over to " + logFile);
+ String rolloverReason = isAppendSupported ? "Append not supported" : "Callback failed";
+ LOG.info(rolloverReason + ". Rolling over to " + logFile);
}
} else {
LOG.info(logFile + " does not exist. Create a new file");
@@ -230,6 +236,7 @@ public class HoodieLogFormatWriter implements HoodieLogFormat.Writer {
}
private void createNewFile() throws IOException {
+ logFileWriteCallback.preLogFileCreate(logFile);
this.output =
fs.create(this.logFile.getPath(), false, bufferSize, replication, WriterBuilder.DEFAULT_SIZE_THRESHOLD, null);
}
@@ -239,7 +246,12 @@ public class HoodieLogFormatWriter implements HoodieLogFormat.Writer {
if (null != shutdownThread) {
Runtime.getRuntime().removeShutdownHook(shutdownThread);
}
- closeStream();
+ logFileWriteCallback.preLogFileClose(logFile);
+ try {
+ closeStream();
+ } finally {
+ logFileWriteCallback.postLogFileClose(logFile);
+ }
}
private void closeStream() throws IOException {
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/marker/MarkerOperation.java b/hudi-common/src/main/java/org/apache/hudi/common/table/marker/MarkerOperation.java
index 81836bdb852..035cf7427b6 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/marker/MarkerOperation.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/marker/MarkerOperation.java
@@ -34,6 +34,7 @@ public class MarkerOperation implements Serializable {
// GET requests
public static final String ALL_MARKERS_URL = String.format("%s/%s", BASE_URL, "all");
public static final String CREATE_AND_MERGE_MARKERS_URL = String.format("%s/%s", BASE_URL, "create-and-merge");
+ public static final String APPEND_MARKERS_URL = String.format("%s/%s", BASE_URL, "append");
public static final String MARKERS_DIR_EXISTS_URL = String.format("%s/%s", BASE_URL, "dir/exists");
// POST requests
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 4254d2aecd3..480ae76a5a1 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
@@ -691,34 +691,6 @@ public class HoodieTableMetadataUtil {
return Collections.singletonMap(MetadataPartitionType.FILES, rollbackRecordsRDD);
}
- private static void reAddLogFilesFromRollbackPlan(HoodieTableMetaClient dataTableMetaClient, String instantTime,
- Map<String, Map<String, Long>> partitionToFilesMap) {
- HoodieInstant rollbackInstant = new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.ROLLBACK_ACTION, instantTime);
- HoodieInstant requested = HoodieTimeline.getRollbackRequestedInstant(rollbackInstant);
- try {
- HoodieRollbackPlan rollbackPlan = TimelineMetadataUtils.deserializeAvroMetadata(
- dataTableMetaClient.getActiveTimeline().readRollbackInfoAsBytes(requested).get(), HoodieRollbackPlan.class);
-
- rollbackPlan.getRollbackRequests().forEach(rollbackRequest -> {
- final String partitionId = getPartitionIdentifierForFilesPartition(rollbackRequest.getPartitionPath());
- partitionToFilesMap.computeIfAbsent(partitionId, s -> new HashMap<>());
- // fetch only log files that are expected to be RB'd in DT as part of this rollback. these log files will not be deleted, but rendered
- // invalid once rollback is complete.
- if (!rollbackRequest.getLogBlocksToBeDeleted().isEmpty()) {
- Map<String, Long> logFiles = new HashMap<>();
- rollbackRequest.getLogBlocksToBeDeleted().forEach((k,v) -> {
- String fileName = k.substring(k.lastIndexOf("/") + 1);
- // rollback plan may not have size for log files to be rolled back. but while merging w/ original commits, the size will get adjusted.
- logFiles.put(fileName, 1L);
- });
- partitionToFilesMap.get(partitionId).putAll(logFiles);
- }
- });
- } catch (IOException e) {
- throw new HoodieMetadataException("Parsing rollback plan for " + rollbackInstant.toString() + " failed ");
- }
- }
-
/**
* Convert rollback action metadata to files partition records.
* Consider only new log files added.
@@ -728,7 +700,6 @@ public class HoodieTableMetadataUtil {
HoodieTableMetaClient dataTableMetaClient) {
Map<String, Map<String, Long>> partitionToAppendedFiles = new HashMap<>();
processRollbackMetadata(rollbackMetadata, partitionToAppendedFiles);
- reAddLogFilesFromRollbackPlan(dataTableMetaClient, instantTime, partitionToAppendedFiles);
return convertFilesToFilesPartitionRecords(Collections.emptyMap(), partitionToAppendedFiles, instantTime, "Rollback");
}
@@ -765,6 +736,12 @@ public class HoodieTableMetadataUtil {
String fileName = new Path(path).getName();
partitionToAppendedFiles.get(partitionId).merge(fileName, size, fileMergeFn);
});
+
+ // Extract original log files from failed commit
+ pm.getLogFilesFromFailedCommit().forEach((path, size) -> {
+ String fileName = new Path(path).getName();
+ partitionToAppendedFiles.get(partitionId).merge(fileName, size, fileMergeFn);
+ });
}
});
}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/data/TestHoodieListDataPairData.java b/hudi-common/src/test/java/org/apache/hudi/common/data/TestHoodieListDataPairData.java
index 9a20fe9bdb2..8355a5f30ed 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/data/TestHoodieListDataPairData.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/data/TestHoodieListDataPairData.java
@@ -224,6 +224,40 @@ public class TestHoodieListDataPairData {
assertEquals(sourceList, originalListData.collectAsList());
}
+ @Test
+ public void testJoin() {
+ // Prepare test data
+ List<Pair<String, String>> leftData = Arrays.asList(
+ Pair.of("a", "value1"),
+ Pair.of("b", "value2"),
+ Pair.of("c", "value3")
+ );
+
+ List<Pair<String, String>> rightData = Arrays.asList(
+ Pair.of("a", "rValue1"),
+ Pair.of("a", "rValue2"),
+ Pair.of("b", "rValue3"),
+ Pair.of("d", "rValue4")
+ );
+
+ HoodiePairData<String, String> left = new HoodieListPairData<>(leftData.stream(), true);
+ HoodiePairData<String, String> right = new HoodieListPairData<>(rightData.stream(), true);
+
+ // Execute the join
+ HoodiePairData<String, Pair<String, String>> joined = left.join(right);
+
+ // Validate the result
+ List<Pair<String, Pair<String, String>>> expected = Arrays.asList(
+ Pair.of("a", Pair.of("value1", "rValue1")),
+ Pair.of("a", Pair.of("value1", "rValue2")),
+ Pair.of("b", Pair.of("value2", "rValue3"))
+ );
+
+ List<Pair<String, Pair<String, String>>> result = joined.collectAsList();
+
+ assertEquals(expected, result, "Join result does not match expected output");
+ }
+
private static List<Pair<String, String>> constructPairs() {
return Arrays.asList(
ImmutablePair.of(KEY1, STRING_VALUE1),
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtils.java
index 75d302dd235..644909125fe 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtils.java
@@ -28,6 +28,7 @@ import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
@@ -54,6 +55,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.TreeSet;
@@ -569,6 +571,25 @@ public class TestFSUtils extends HoodieCommonTestHarness {
FSUtils.makeQualified(wrapperStorage, new StoragePath("s3://x/y")));
}
+ @Test
+ public void testGetFileStatusesUnderPartition() throws IOException {
+ Path hoodieTempDir = getHoodieTempDir();
+ FileSystem fileSystem = metaClient.getFs();
+ prepareTestDirectory(fileSystem, hoodieTempDir);
+ List<Option<FileStatus>> fileStatusList = FSUtils.getFileStatusesUnderPartition(
+ fileSystem,
+ new Path(baseUri.toString(), ".hoodie/.temp"),
+ new HashSet<>(Collections.singletonList("file3.txt")),
+ false);
+ assertEquals(1, fileStatusList.size());
+
+ assertThrows(HoodieIOException.class, () -> FSUtils.getFileStatusesUnderPartition(
+ fileSystem,
+ new Path(baseUri.toString(), ".hoodie/.temp"),
+ new HashSet<>(Collections.singletonList("file4.txt")),
+ false));
+ }
+
private Path getHoodieTempDir() {
return new Path(baseUri.toString(), ".hoodie/.temp");
}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsMocked.java b/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsMocked.java
new file mode 100644
index 00000000000..e60f9c6a0a9
--- /dev/null
+++ b/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsMocked.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.fs;
+
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieIOException;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class TestFSUtilsMocked {
+
+ @Mock
+ private FileSystem mockFileSystem;
+
+ private final Path basePath = new Path("/base/path");
+ private final Set<String> fileNames = new HashSet<>(Arrays.asList("file1.txt", "file2.txt"));
+ private FileStatus mockFileStatus1;
+ private FileStatus mockFileStatus2;
+
+ @BeforeEach
+ public void setUp() {
+ MockitoAnnotations.initMocks(this);
+ mockFileStatus1 = new FileStatus(100, false, 3, 1024, 0, new Path("/base/path/file1.txt"));
+ mockFileStatus2 = new FileStatus(200, false, 3, 1024, 0, new Path("/base/path/file2.txt"));
+ }
+
+ @Test
+ public void testGetFileStatusesUnderPartitionWithListStatus() throws IOException, IOException {
+ // Setup
+ when(mockFileSystem.getScheme()).thenReturn("file"); // Assuming "file" is list status friendly
+ when(mockFileSystem.listStatus(eq(basePath), any())).thenReturn(new FileStatus[] {mockFileStatus1, mockFileStatus2});
+
+ // Execute
+ List<Option<FileStatus>> result = FSUtils.getFileStatusesUnderPartition(mockFileSystem, basePath, fileNames, false);
+
+ // Verify
+ assertEquals(2, result.size());
+ assertTrue(result.get(0).isPresent());
+ assertTrue(result.get(1).isPresent());
+
+ // Cleanup
+ verify(mockFileSystem, times(1)).listStatus((Path) any(), any());
+ }
+
+ @Test
+ public void testGetFileStatusesUnderPartitionIgnoringMissingFiles() throws IOException {
+ // Setup for scenario where file2.txt does not exist
+ when(mockFileSystem.getScheme()).thenReturn("hdfs"); // Assuming "hdfs" is not list status friendly
+ when(mockFileSystem.getFileStatus(new Path("/base/path/file1.txt"))).thenReturn(mockFileStatus1);
+ when(mockFileSystem.getFileStatus(new Path("/base/path/file2.txt"))).thenThrow(new FileNotFoundException());
+
+ // Execute
+ List<Option<FileStatus>> result = FSUtils.getFileStatusesUnderPartition(mockFileSystem, basePath, fileNames, true);
+
+ // Verify
+ assertEquals(2, result.size());
+ assertTrue(result.get(0).isPresent());
+ assertFalse(result.get(1).isPresent()); // Missing file results in an empty Option
+
+ // Cleanup
+ verify(mockFileSystem, times(2)).getFileStatus(any());
+ }
+
+ @Test
+ public void testGetFileStatusesUnderPartitionThrowsHoodieIOException() throws IOException {
+ // Setup
+ when(mockFileSystem.getScheme()).thenReturn("file"); // Assuming "file" is list status friendly
+ when(mockFileSystem.listStatus((Path) any(), any())).thenThrow(new IOException());
+
+ // Execute & Verify
+ assertThrows(HoodieIOException.class, () ->
+ FSUtils.getFileStatusesUnderPartition(mockFileSystem, basePath, fileNames, false));
+
+ // Cleanup
+ verify(mockFileSystem, times(1)).listStatus((Path) any(), any());
+ }
+}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java
index c3008fd171a..82f6a8c9f75 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java
@@ -30,6 +30,7 @@ import org.apache.hudi.avro.model.HoodieSavepointMetadata;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.model.HoodieLogFile;
import org.apache.hudi.common.model.HoodiePartitionMetadata;
import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
import org.apache.hudi.common.model.IOType;
@@ -96,17 +97,20 @@ public class FileCreateUtils {
return FSUtils.makeLogFileName(fileId, fileExtension, instantTime, version, WRITE_TOKEN);
}
- public static String markerFileName(String instantTime, String fileId, IOType ioType) {
- return markerFileName(instantTime, fileId, ioType, BASE_FILE_EXTENSION);
+ public static String markerFileName(String fileName, IOType ioType) {
+ return String.format("%s%s.%s", fileName, HoodieTableMetaClient.MARKER_EXTN, ioType.name());
}
- public static String markerFileName(String instantTime, String fileId, IOType ioType, String fileExtension) {
- return markerFileName(instantTime, fileId, ioType, fileExtension, WRITE_TOKEN);
+ public static String dataFileMarkerFileName(String instantTime, String fileId, IOType ioType, String fileExtension, String writeToken) {
+ return markerFileName(FSUtils.makeBaseFileName(instantTime, writeToken, fileId, fileExtension), ioType);
}
- public static String markerFileName(String instantTime, String fileId, IOType ioType, String fileExtension, String writeToken) {
- return String.format("%s_%s_%s%s%s.%s", fileId, writeToken, instantTime, fileExtension,
- HoodieTableMetaClient.MARKER_EXTN, ioType);
+ public static String logFileMarkerFileName(String instantTime, String fileId, IOType ioType, int logVersion) {
+ return logFileMarkerFileName(instantTime, fileId, ioType, HoodieLogFile.DELTA_EXTENSION, logVersion);
+ }
+
+ public static String logFileMarkerFileName(String instantTime, String fileId, IOType ioType, String fileExtension, int logVersion) {
+ return markerFileName(FSUtils.makeLogFileName(fileId, fileExtension, instantTime, logVersion, WRITE_TOKEN), ioType);
}
private static void createMetaFile(String basePath, String instantTime, String suffix, FileSystem fs) throws IOException {
@@ -368,9 +372,36 @@ public class FileCreateUtils {
public static String createMarkerFile(String basePath, String partitionPath, String commitInstant,
String instantTime, String fileId, IOType ioType, String writeToken) throws IOException {
- Path parentPath = Paths.get(basePath, HoodieTableMetaClient.TEMPFOLDER_NAME, commitInstant, partitionPath);
+ Path parentPath = Paths.get(basePath, HoodieTableMetaClient.TEMPFOLDER_NAME, instantTime, partitionPath);
+ Files.createDirectories(parentPath);
+ Path markerFilePath = parentPath.resolve(dataFileMarkerFileName(instantTime, fileId, ioType, BASE_FILE_EXTENSION, writeToken));
+ if (Files.notExists(markerFilePath)) {
+ Files.createFile(markerFilePath);
+ }
+ return markerFilePath.toAbsolutePath().toString();
+ }
+
+ public static String createLogFileMarker(String basePath, String partitionPath, String instantTime, String fileId, IOType ioType)
+ throws IOException {
+ return createLogFileMarker(basePath, partitionPath, instantTime, fileId, ioType, HoodieLogFile.LOGFILE_BASE_VERSION);
+ }
+
+ public static String createLogFileMarker(String basePath, String partitionPath, String instantTime, String fileId, IOType ioType, int logVersion)
+ throws IOException {
+ Path parentPath = Paths.get(basePath, HoodieTableMetaClient.TEMPFOLDER_NAME, instantTime, partitionPath);
+ Files.createDirectories(parentPath);
+ Path markerFilePath = parentPath.resolve(logFileMarkerFileName(instantTime, fileId, ioType, logVersion));
+ if (Files.notExists(markerFilePath)) {
+ Files.createFile(markerFilePath);
+ }
+ return markerFilePath.toAbsolutePath().toString();
+ }
+
+ public static String createFileMarkerByFileName(String basePath, String partitionPath, String instantTime, String fileName, IOType ioType)
+ throws IOException {
+ Path parentPath = Paths.get(basePath, HoodieTableMetaClient.TEMPFOLDER_NAME, instantTime, partitionPath);
Files.createDirectories(parentPath);
- Path markerFilePath = parentPath.resolve(markerFileName(instantTime, fileId, ioType, BASE_FILE_EXTENSION, writeToken));
+ Path markerFilePath = parentPath.resolve(markerFileName(fileName, ioType));
if (Files.notExists(markerFilePath)) {
Files.createFile(markerFilePath);
}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java
index db40a271a6d..b78665644fb 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java
@@ -108,6 +108,7 @@ import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightRep
import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightRollbackFile;
import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightSavepoint;
import static org.apache.hudi.common.testutils.FileCreateUtils.createMarkerFile;
+import static org.apache.hudi.common.testutils.FileCreateUtils.createLogFileMarker;
import static org.apache.hudi.common.testutils.FileCreateUtils.createReplaceCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createRequestedCleanFile;
import static org.apache.hudi.common.testutils.FileCreateUtils.createRequestedCommit;
@@ -598,6 +599,11 @@ public class HoodieTestTable {
return this;
}
+ public HoodieTestTable withLogMarkerFile(String partitionPath, String fileId, IOType ioType) throws IOException {
+ createLogFileMarker(basePath, partitionPath, currentInstantTime, fileId, ioType);
+ return this;
+ }
+
/**
* Insert one base file to each of the given distinct partitions.
*
@@ -776,6 +782,7 @@ public class HoodieTestTable {
public FileStatus[] listAllLogFiles(String fileExtension) throws IOException {
return FileSystemTestUtils.listRecursive(fs, new Path(basePath)).stream()
+ .filter(status -> !status.getPath().toString().contains(HoodieTableMetaClient.METAFOLDER_NAME))
.filter(status -> status.getPath().getName().contains(fileExtension))
.toArray(FileStatus[]::new);
}
@@ -1064,7 +1071,7 @@ public class HoodieTestTable {
return commitMetadata;
}
- private Option<HoodieCommitMetadata> getMetadataForInstant(String instantTime) {
+ public Option<HoodieCommitMetadata> getMetadataForInstant(String instantTime) {
metaClient = HoodieTableMetaClient.reload(metaClient);
Option<HoodieInstant> hoodieInstant = metaClient.getActiveTimeline().getCommitsTimeline()
.filterCompletedInstants().filter(i -> i.getTimestamp().equals(instantTime)).firstInstant();
diff --git a/hudi-io/src/main/java/org/apache/hudi/storage/StorageSchemes.java b/hudi-io/src/main/java/org/apache/hudi/storage/StorageSchemes.java
index 371d31ac95d..129956166b3 100644
--- a/hudi-io/src/main/java/org/apache/hudi/storage/StorageSchemes.java
+++ b/hudi-io/src/main/java/org/apache/hudi/storage/StorageSchemes.java
@@ -26,62 +26,63 @@ import java.util.Arrays;
*/
public enum StorageSchemes {
// Local filesystem
- FILE("file", false, false, true),
+ FILE("file", false, false, true, true),
// Hadoop File System
- HDFS("hdfs", true, false, true),
+ HDFS("hdfs", true, false, true, false),
// Baidu Advanced File System
- AFS("afs", true, null, null),
+ AFS("afs", true, null, null, null),
// Mapr File System
- MAPRFS("maprfs", true, null, null),
+ MAPRFS("maprfs", true, null, null, null),
// Apache Ignite FS
- IGNITE("igfs", true, null, null),
+ IGNITE("igfs", true, null, null, null),
// AWS S3
- S3A("s3a", false, true, null), S3("s3", false, true, null),
+ S3A("s3a", false, true, null, true),
+ S3("s3", false, true, null, true),
// Google Cloud Storage
- GCS("gs", false, true, null),
+ GCS("gs", false, true, null, true),
// Azure WASB
- WASB("wasb", false, null, null), WASBS("wasbs", false, null, null),
+ WASB("wasb", false, null, null, null), WASBS("wasbs", false, null, null, null),
// Azure ADLS
- ADL("adl", false, null, null),
+ ADL("adl", false, null, null, null),
// Azure ADLS Gen2
- ABFS("abfs", false, null, null), ABFSS("abfss", false, null, null),
+ ABFS("abfs", false, null, null, null), ABFSS("abfss", false, null, null, null),
// Aliyun OSS
- OSS("oss", false, null, null),
+ OSS("oss", false, null, null, null),
// View FS for federated setups. If federating across cloud stores, then append support is false
// View FS support atomic creation
- VIEWFS("viewfs", true, null, true),
+ VIEWFS("viewfs", true, null, true, null),
//ALLUXIO
- ALLUXIO("alluxio", false, null, null),
+ ALLUXIO("alluxio", false, null, null, null),
// Tencent Cloud Object Storage
- COSN("cosn", false, null, null),
+ COSN("cosn", false, null, null, null),
// Tencent Cloud HDFS
- CHDFS("ofs", true, null, null),
+ CHDFS("ofs", true, null, null, null),
// Tencent Cloud CacheFileSystem
- GOOSEFS("gfs", false, null, null),
+ GOOSEFS("gfs", false, null, null, null),
// Databricks file system
- DBFS("dbfs", false, null, null),
+ DBFS("dbfs", false, null, null, null),
// IBM Cloud Object Storage
- COS("cos", false, null, null),
+ COS("cos", false, null, null, null),
// Huawei Cloud Object Storage
- OBS("obs", false, null, null),
+ OBS("obs", false, null, null, null),
// Kingsoft Standard Storage ks3
- KS3("ks3", false, null, null),
+ KS3("ks3", false, null, null, null),
// Netease Object Storage nos
- NOS("nos", false, null, null),
+ NOS("nos", false, null, null, null),
// JuiceFileSystem
- JFS("jfs", true, null, null),
+ JFS("jfs", true, null, null, null),
// Baidu Object Storage
- BOS("bos", false, null, null),
+ BOS("bos", false, null, null, null),
// Oracle Cloud Infrastructure Object Storage
- OCI("oci", false, null, null),
+ OCI("oci", false, null, null, null),
// Volcengine Object Storage
- TOS("tos", false, null, null),
+ TOS("tos", false, null, null, null),
// Volcengine Cloud HDFS
- CFS("cfs", true, null, null),
+ CFS("cfs", true, null, null, null),
// Aliyun Apsara File Storage for HDFS
- DFS("dfs", true, false, true),
+ DFS("dfs", true, false, true, null),
// Hopsworks File System
- HOPSFS("hopsfs", false, false, true);
+ HOPSFS("hopsfs", false, false, true, null);
private String scheme;
private boolean supportsAppend;
@@ -89,12 +90,17 @@ public enum StorageSchemes {
private Boolean isWriteTransactional;
// null for uncertain if dfs support atomic create&delete, please update this for each FS
private Boolean supportAtomicCreation;
+ // list files may bring pressure to storage with centralized meta service like HDFS.
+ // when we want to get only part of files under a directory rather than all files, use getStatus may be more friendly than listStatus.
+ // here is a trade-off between rpc times and throughput of storage meta service
+ private Boolean listStatusFriendly;
- StorageSchemes(String scheme, boolean supportsAppend, Boolean isWriteTransactional, Boolean supportAtomicCreation) {
+ StorageSchemes(String scheme, boolean supportsAppend, Boolean isWriteTransactional, Boolean supportAtomicCreation, Boolean listStatusFriendly) {
this.scheme = scheme;
this.supportsAppend = supportsAppend;
this.isWriteTransactional = isWriteTransactional;
this.supportAtomicCreation = supportAtomicCreation;
+ this.listStatusFriendly = listStatusFriendly;
}
public String getScheme() {
@@ -113,6 +119,10 @@ public enum StorageSchemes {
return supportAtomicCreation != null && supportAtomicCreation;
}
+ public boolean getListStatusFriendly() {
+ return listStatusFriendly != null && listStatusFriendly;
+ }
+
public static boolean isSchemeSupported(String scheme) {
return Arrays.stream(values()).anyMatch(s -> s.getScheme().equals(scheme));
}
@@ -138,4 +148,11 @@ public enum StorageSchemes {
}
return Arrays.stream(StorageSchemes.values()).anyMatch(s -> s.isAtomicCreationSupported() && s.scheme.equals(scheme));
}
+
+ public static boolean isListStatusFriendly(String scheme) {
+ if (!isSchemeSupported(scheme)) {
+ throw new IllegalArgumentException("Unsupported scheme :" + scheme);
+ }
+ return Arrays.stream(StorageSchemes.values()).anyMatch(s -> s.getListStatusFriendly() && s.scheme.equals(scheme));
+ }
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCallProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCallProcedure.scala
index 30bec0f8a9c..14e6a595f57 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCallProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCallProcedure.scala
@@ -196,7 +196,7 @@ class TestCallProcedure extends HoodieSparkProcedureTestBase {
s"Argument: instant_time is required")
val instantTime = "101"
- FileCreateUtils.createMarkerFile(tablePath, "", instantTime, "f0", IOType.APPEND)
+ FileCreateUtils.createLogFileMarker(tablePath, "", instantTime, "f0", IOType.APPEND)
assertResult(1) {
FileCreateUtils.getTotalMarkerFileCount(tablePath, "", instantTime, IOType.APPEND)
}
@@ -234,12 +234,12 @@ class TestCallProcedure extends HoodieSparkProcedureTestBase {
s"Argument: instant_time is required")
var instantTime = "101"
- FileCreateUtils.createMarkerFile(tablePath, "", instantTime, "f0", IOType.APPEND)
+ FileCreateUtils.createLogFileMarker(tablePath, "", instantTime, "f0", IOType.APPEND)
assertResult(1) {
FileCreateUtils.getTotalMarkerFileCount(tablePath, "", instantTime, IOType.APPEND)
}
instantTime = "102"
- FileCreateUtils.createMarkerFile(tablePath, "", instantTime, "f0", IOType.APPEND)
+ FileCreateUtils.createLogFileMarker(tablePath, "", instantTime, "f0", IOType.APPEND)
assertResult(1) {
FileCreateUtils.getTotalMarkerFileCount(tablePath, "", instantTime, IOType.APPEND)
}
diff --git a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java
index 08b4e903a66..24e9d06018e 100644
--- a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java
+++ b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/RequestHandler.java
@@ -498,6 +498,13 @@ public class RequestHandler {
writeValueAsString(ctx, markers);
}, false));
+ app.get(MarkerOperation.APPEND_MARKERS_URL, new ViewHandler(ctx -> {
+ metricsRegistry.add("APPEND_MARKERS", 1);
+ Set<String> markers = markerHandler.getAppendMarkers(
+ ctx.queryParamAsClass(MarkerOperation.MARKER_DIR_PATH_PARAM, String.class).getOrDefault(""));
+ writeValueAsString(ctx, markers);
+ }, false));
+
app.get(MarkerOperation.MARKERS_DIR_EXISTS_URL, new ViewHandler(ctx -> {
metricsRegistry.add("MARKERS_DIR_EXISTS", 1);
boolean exist = markerHandler.doesMarkerDirExist(
diff --git a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/MarkerHandler.java b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/MarkerHandler.java
index 42e2f40e629..620ea852539 100644
--- a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/MarkerHandler.java
+++ b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/MarkerHandler.java
@@ -162,6 +162,16 @@ public class MarkerHandler extends Handler {
.collect(Collectors.toSet());
}
+ /**
+ * @param markerDir marker directory path
+ * @return all marker paths of write IO type "APPEND"
+ */
+ public Set<String> getAppendMarkers(String markerDir) {
+ return getAllMarkers(markerDir).stream()
+ .filter(markerName -> markerName.endsWith(IOType.APPEND.name()))
+ .collect(Collectors.toSet());
+ }
+
/**
* @param markerDir marker directory path
* @return {@code true} if the marker directory exists; {@code false} otherwise.