You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2022/02/03 22:01:26 UTC
[GitHub] [hudi] yihua commented on a change in pull request #4556: [HUDI-3191] Removing duplicating file-listing process w/in Hive's MOR `FileInputFormat`s
yihua commented on a change in pull request #4556:
URL: https://github.com/apache/hudi/pull/4556#discussion_r799001688
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieFileInputFormatBase.java
##########
@@ -143,6 +121,127 @@ private static FileStatus getFileStatusUnchecked(Option<HoodieBaseFile> baseFile
return returns.toArray(new FileStatus[0]);
}
+ private void validate(List<FileStatus> targetFiles, List<FileStatus> legacyFileStatuses) {
+ List<FileStatus> diff = CollectionUtils.diff(targetFiles, legacyFileStatuses);
+ checkState(diff.isEmpty(), "Should be empty");
+ }
+
+ @Nonnull
+ private static FileStatus getFileStatusUnchecked(HoodieBaseFile baseFile) {
+ try {
+ return HoodieInputFormatUtils.getFileStatus(baseFile);
+ } catch (IOException ioe) {
+ throw new HoodieIOException("Failed to get file-status", ioe);
+ }
+ }
+
+ /**
+ * Abstracts and exposes {@link FileInputFormat#listStatus(JobConf)} operation to subclasses that
+ * lists files (returning an array of {@link FileStatus}) corresponding to the input paths specified
+ * as part of provided {@link JobConf}
+ */
+ protected final FileStatus[] doListStatus(JobConf job) throws IOException {
+ return super.listStatus(job);
+ }
+
+ /**
+ * Achieves listStatus functionality for an incrementally queried table. Instead of listing all
+ * partitions and then filtering based on the commits of interest, this logic first extracts the
+ * partitions touched by the desired commits and then lists only those partitions.
+ */
+ protected List<FileStatus> listStatusForIncrementalMode(JobConf job,
+ HoodieTableMetaClient tableMetaClient,
+ List<Path> inputPaths,
+ String incrementalTable) throws IOException {
+ String tableName = tableMetaClient.getTableConfig().getTableName();
+ Job jobContext = Job.getInstance(job);
+ Option<HoodieTimeline> timeline = HoodieInputFormatUtils.getFilteredCommitsTimeline(jobContext, tableMetaClient);
+ if (!timeline.isPresent()) {
+ return null;
+ }
+ Option<List<HoodieInstant>> commitsToCheck = HoodieInputFormatUtils.getCommitsForIncrementalQuery(jobContext, tableName, timeline.get());
+ if (!commitsToCheck.isPresent()) {
+ return null;
+ }
+ Option<String> incrementalInputPaths = HoodieInputFormatUtils.getAffectedPartitions(commitsToCheck.get(), tableMetaClient, timeline.get(), inputPaths);
+ // Mutate the JobConf to set the input paths to only partitions touched by incremental pull.
+ if (!incrementalInputPaths.isPresent()) {
+ return null;
+ }
+ setInputPaths(job, incrementalInputPaths.get());
+ FileStatus[] fileStatuses = doListStatus(job);
+ return HoodieInputFormatUtils.filterIncrementalFileStatus(jobContext, tableMetaClient, timeline.get(), fileStatuses, commitsToCheck.get());
+ }
+
+ protected abstract boolean includeLogFilesForSnapshotView();
+
+ @Nonnull
+ private static RealtimeFileStatus createRealtimeFileStatusUnchecked(HoodieBaseFile baseFile,
+ Stream<HoodieLogFile> logFiles,
+ Option<HoodieInstant> latestCompletedInstantOpt,
+ HoodieTableMetaClient tableMetaClient) {
+ List<HoodieLogFile> sortedLogFiles = logFiles.sorted(HoodieLogFile.getLogFileComparator()).collect(Collectors.toList());
+ FileStatus baseFileStatus = getFileStatusUnchecked(baseFile);
+ try {
+ RealtimeFileStatus rtFileStatus = new RealtimeFileStatus(baseFileStatus);
+ rtFileStatus.setDeltaLogFiles(sortedLogFiles);
+ rtFileStatus.setBaseFilePath(baseFile.getPath());
+ rtFileStatus.setBasePath(tableMetaClient.getBasePath());
+
+ if (latestCompletedInstantOpt.isPresent()) {
+ HoodieInstant latestCompletedInstant = latestCompletedInstantOpt.get();
+ checkState(latestCompletedInstant.isCompleted());
+
+ rtFileStatus.setMaxCommitTime(latestCompletedInstant.getTimestamp());
+ }
+
+ if (baseFileStatus instanceof LocatedFileStatusWithBootstrapBaseFile || baseFileStatus instanceof FileStatusWithBootstrapBaseFile) {
+ rtFileStatus.setBootStrapFileStatus(baseFileStatus);
+ }
+
+ return rtFileStatus;
+ } catch (IOException e) {
+ throw new HoodieIOException(String.format("Failed to init %s", RealtimeFileStatus.class.getSimpleName()), e);
+ }
+ }
+
+ @Nonnull
+ private List<FileStatus> listStatusForSnapshotModeLegacy(JobConf job, Map<String, HoodieTableMetaClient> tableMetaClientMap, List<Path> snapshotPaths) throws IOException {
+ return HoodieInputFormatUtils.filterFileStatusForSnapshotMode(job, tableMetaClientMap, snapshotPaths, includeLogFilesForSnapshotView());
+ }
+
+ @Nonnull
+ private static RealtimeFileStatus createRealtimeFileStatusUnchecked(HoodieLogFile latestLogFile,
Review comment:
Discussed offline. Mentioned above.
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/functional/TestHoodieSparkMergeOnReadTableRollback.java
##########
@@ -516,8 +556,6 @@ void testMORTableRestore(boolean restoreAfterCompaction) throws Exception {
JavaRDD<HoodieRecord> writeRecords = jsc().parallelize(records, 1);
JavaRDD<WriteStatus> writeStatusJavaRDD = client.upsert(writeRecords, newCommitTime);
client.commit(newCommitTime, writeStatusJavaRDD);
- List<WriteStatus> statuses = writeStatusJavaRDD.collect();
- assertNoWriteErrors(statuses);
Review comment:
Makes sense
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/HoodieFileInputFormatBase.java
##########
@@ -143,6 +121,127 @@ private static FileStatus getFileStatusUnchecked(Option<HoodieBaseFile> baseFile
return returns.toArray(new FileStatus[0]);
}
+ private void validate(List<FileStatus> targetFiles, List<FileStatus> legacyFileStatuses) {
+ List<FileStatus> diff = CollectionUtils.diff(targetFiles, legacyFileStatuses);
+ checkState(diff.isEmpty(), "Should be empty");
+ }
+
+ @Nonnull
+ private static FileStatus getFileStatusUnchecked(HoodieBaseFile baseFile) {
+ try {
+ return HoodieInputFormatUtils.getFileStatus(baseFile);
+ } catch (IOException ioe) {
+ throw new HoodieIOException("Failed to get file-status", ioe);
+ }
+ }
+
+ /**
+ * Abstracts and exposes {@link FileInputFormat#listStatus(JobConf)} operation to subclasses that
+ * lists files (returning an array of {@link FileStatus}) corresponding to the input paths specified
+ * as part of provided {@link JobConf}
+ */
+ protected final FileStatus[] doListStatus(JobConf job) throws IOException {
+ return super.listStatus(job);
+ }
+
+ /**
+ * Achieves listStatus functionality for an incrementally queried table. Instead of listing all
+ * partitions and then filtering based on the commits of interest, this logic first extracts the
+ * partitions touched by the desired commits and then lists only those partitions.
+ */
+ protected List<FileStatus> listStatusForIncrementalMode(JobConf job,
+ HoodieTableMetaClient tableMetaClient,
+ List<Path> inputPaths,
+ String incrementalTable) throws IOException {
+ String tableName = tableMetaClient.getTableConfig().getTableName();
+ Job jobContext = Job.getInstance(job);
+ Option<HoodieTimeline> timeline = HoodieInputFormatUtils.getFilteredCommitsTimeline(jobContext, tableMetaClient);
+ if (!timeline.isPresent()) {
+ return null;
+ }
+ Option<List<HoodieInstant>> commitsToCheck = HoodieInputFormatUtils.getCommitsForIncrementalQuery(jobContext, tableName, timeline.get());
+ if (!commitsToCheck.isPresent()) {
+ return null;
+ }
+ Option<String> incrementalInputPaths = HoodieInputFormatUtils.getAffectedPartitions(commitsToCheck.get(), tableMetaClient, timeline.get(), inputPaths);
+ // Mutate the JobConf to set the input paths to only partitions touched by incremental pull.
+ if (!incrementalInputPaths.isPresent()) {
+ return null;
+ }
+ setInputPaths(job, incrementalInputPaths.get());
+ FileStatus[] fileStatuses = doListStatus(job);
+ return HoodieInputFormatUtils.filterIncrementalFileStatus(jobContext, tableMetaClient, timeline.get(), fileStatuses, commitsToCheck.get());
+ }
+
+ protected abstract boolean includeLogFilesForSnapshotView();
+
+ @Nonnull
+ private static RealtimeFileStatus createRealtimeFileStatusUnchecked(HoodieBaseFile baseFile,
Review comment:
As discussed offline, one-line javadocs are going to be added for both methods for clarity in the following PR:
```
/**
* Creates real-time FileStatus for a base file with log files.
*/
@Nonnull
private static RealtimeFileStatus createRealtimeFileStatusUnchecked(HoodieBaseFile baseFile,
Stream<HoodieLogFile> logFiles,
Option<HoodieInstant> latestCompletedInstantOpt,
HoodieTableMetaClient tableMetaClient)
/**
* Creates real-time FileStatus for the log files only.
*/
@Nonnull
private static RealtimeFileStatus createRealtimeFileStatusUnchecked(HoodieLogFile latestLogFile,
Stream<HoodieLogFile> logFiles,
Option<HoodieInstant> latestCompletedInstantOpt,
HoodieTableMetaClient tableMetaClient)
```
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieRealtimeFileSplit.java
##########
@@ -44,9 +44,7 @@
private Option<HoodieVirtualKeyInfo> hoodieVirtualKeyInfo = Option.empty();
- public HoodieRealtimeFileSplit() {
- super();
Review comment:
Got it.
##########
File path: hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestHoodieMergeOnReadTable.java
##########
@@ -189,8 +190,10 @@ public void testUpsertPartitioner(boolean populateMetaFields) throws Exception {
assertTrue(fileIdToNewSize.entrySet().stream().anyMatch(entry -> fileIdToSize.get(entry.getKey()) < entry.getValue()));
- List<String> dataFiles = roView.getLatestBaseFiles().map(HoodieBaseFile::getPath).collect(Collectors.toList());
- List<GenericRecord> recordsRead = HoodieMergeOnReadTestUtils.getRecordsUsingInputFormat(hadoopConf(), dataFiles,
+ List<String> inputPaths = roView.getLatestBaseFiles()
+ .map(baseFile -> new Path(baseFile.getPath()).getParent().toString())
+ .collect(Collectors.toList());
Review comment:
Discussed offline that the inputPaths are going to be kept as is in the tests.
##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeInputFormatUtils.java
##########
@@ -65,11 +65,71 @@
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static org.apache.hudi.TypeUtils.unsafeCast;
+import static org.apache.hudi.common.util.ValidationUtils.checkState;
+
public class HoodieRealtimeInputFormatUtils extends HoodieInputFormatUtils {
private static final Logger LOG = LogManager.getLogger(HoodieRealtimeInputFormatUtils.class);
- public static InputSplit[] getRealtimeSplits(Configuration conf, Stream<FileSplit> fileSplits) {
+ public static InputSplit[] getRealtimeSplits(Configuration conf, List<FileSplit> fileSplits) throws IOException {
+ if (fileSplits.isEmpty()) {
+ return new InputSplit[0];
+ }
+
+ FileSplit fileSplit = fileSplits.get(0);
+
+ // Pre-process table-config to fetch virtual key info
+ Path partitionPath = fileSplit.getPath().getParent();
+ HoodieTableMetaClient metaClient = getTableMetaClientForBasePathUnchecked(conf, partitionPath);
+
+ Option<HoodieVirtualKeyInfo> hoodieVirtualKeyInfoOpt = getHoodieVirtualKeyInfo(metaClient);
+
+ // NOTE: This timeline is kept in sync w/ {@code HoodieTableFileIndexBase}
+ HoodieInstant latestCommitInstant =
+ metaClient.getActiveTimeline().getCommitsTimeline().filterCompletedInstants().lastInstant().get();
+
+ InputSplit[] finalSplits = fileSplits.stream()
+ .map(split -> {
+ // There are 4 types of splits could we have to handle here
+ // - {@code BootstrapBaseFileSplit}: in case base file does have associated bootstrap file,
+ // but does NOT have any log files appended (convert it to {@code RealtimeBootstrapBaseFileSplit})
+ // - {@code RealtimeBootstrapBaseFileSplit}: in case base file does have associated bootstrap file
+ // and does have log files appended
+ // - {@code BaseFileWithLogsSplit}: in case base file does NOT have associated bootstrap file
Review comment:
I see. Yeah then it's fine, no need to update docs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org