[HUDI-3206] Unify Hive's MOR implementations to avoid duplication #4559
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -21,10 +21,12 @@ | |
| import org.apache.hadoop.conf.Configurable; | ||
| 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 org.apache.hadoop.io.ArrayWritable; | ||
| import org.apache.hadoop.io.NullWritable; | ||
| import org.apache.hadoop.mapred.FileInputFormat; | ||
| import org.apache.hadoop.mapred.FileSplit; | ||
| import org.apache.hadoop.mapred.JobConf; | ||
| import org.apache.hadoop.mapreduce.Job; | ||
| import org.apache.hudi.common.config.TypedProperties; | ||
|
|
@@ -65,12 +67,32 @@ | |
| * <li>Incremental mode: reading table's state as of particular timestamp (or instant, in Hudi's terms)</li> | ||
| * <li>External mode: reading non-Hudi partitions</li> | ||
| * </ul> | ||
| * | ||
| * NOTE: This class is invariant of the underlying file-format of the files being read | ||
| */ | ||
| public abstract class HoodieFileInputFormatBase extends FileInputFormat<NullWritable, ArrayWritable> | ||
| implements Configurable { | ||
|
|
||
| protected Configuration conf; | ||
|
|
||
| @Nonnull | ||
| private static RealtimeFileStatus createRealtimeFileStatusUnchecked(HoodieBaseFile baseFile, Stream<HoodieLogFile> logFiles) { | ||
| 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()); | ||
| if (baseFileStatus instanceof LocatedFileStatusWithBootstrapBaseFile || baseFileStatus instanceof FileStatusWithBootstrapBaseFile) { | ||
| rtFileStatus.setBootStrapFileStatus(baseFileStatus); | ||
nsivabalan marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| } | ||
|
|
||
| return rtFileStatus; | ||
| } catch (IOException e) { | ||
| throw new HoodieIOException(String.format("Failed to init %s", RealtimeFileStatus.class.getSimpleName()), e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public final Configuration getConf() { | ||
| return conf; | ||
|
|
@@ -81,6 +103,24 @@ public final void setConf(Configuration conf) { | |
| this.conf = conf; | ||
| } | ||
|
|
||
| protected abstract boolean includeLogFilesForSnapshotView(); | ||
|
|
||
| @Override | ||
| protected boolean isSplitable(FileSystem fs, Path filename) { | ||
| return !(filename instanceof PathWithBootstrapFileStatus); | ||
| } | ||
|
|
||
| @Override | ||
| protected FileSplit makeSplit(Path file, long start, long length, | ||
| String[] hosts) { | ||
| FileSplit split = new FileSplit(file, start, length, hosts); | ||
|
|
||
| if (file instanceof PathWithBootstrapFileStatus) { | ||
| return makeExternalFileSplit((PathWithBootstrapFileStatus)file, split); | ||
| } | ||
| return split; | ||
| } | ||
|
|
||
| @Override | ||
| public FileStatus[] listStatus(JobConf job) throws IOException { | ||
| // Segregate inputPaths[] to incremental, snapshot and non hoodie paths | ||
|
|
@@ -121,20 +161,6 @@ public FileStatus[] listStatus(JobConf job) throws IOException { | |
| return returns.toArray(new FileStatus[0]); | ||
| } | ||
|
|
||
| /** | ||
| * 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 | ||
|
|
@@ -172,35 +198,25 @@ protected List<FileStatus> listStatusForIncrementalMode(JobConf job, | |
| return HoodieInputFormatUtils.filterIncrementalFileStatus(jobContext, tableMetaClient, timeline.get(), fileStatuses, commitsToCheck.get()); | ||
| } | ||
|
|
||
| @Override | ||
| protected FileSplit makeSplit(Path file, long start, long length, | ||
|
Contributor
There was a problem hiding this comment. nit: put two
Contributor
Author
There was a problem hiding this comment. Will clean this up in the final PR #4743 |
||
| String[] hosts, String[] inMemoryHosts) { | ||
| FileSplit split = new FileSplit(file, start, length, hosts, inMemoryHosts); | ||
| if (file instanceof PathWithBootstrapFileStatus) { | ||
|
Contributor
There was a problem hiding this comment. can you help me understand why don't I see diff FileStatuses here like RealtimeFileStatus for eg, but just PathWithBootstrapFileStatus?
Contributor
There was a problem hiding this comment. My understanding is that these methods are moved from HoodieParquetInputFormat. @alexeykudinkin is the logic parquet agnostic and going to be reused by other file formats like HFile? I think if that's not the case, better to keep them in HoodieParquetInputFormat, since this base class is going to be general for different formats.
Contributor
Author
There was a problem hiding this comment. This is used only for COW
Contributor
There was a problem hiding this comment. @alexeykudinkin what about my above question regarding whether the logic should reside in HoodieParquetInputFormat or this class?
Contributor
Author
There was a problem hiding this comment. @yihua GH comments are weird, when i was responding to @nsivabalan there were no comments of yours, and then it sandwiched your comment in b/w of those.
Your understanding is correct, all of this logic is file-format agnostic. |
||
| return makeExternalFileSplit((PathWithBootstrapFileStatus)file, split); | ||
| } | ||
| return split; | ||
| } | ||
|
|
||
| private BootstrapBaseFileSplit makeExternalFileSplit(PathWithBootstrapFileStatus file, FileSplit split) { | ||
| try { | ||
| LOG.info("Making external data split for " + file); | ||
| FileStatus externalFileStatus = file.getBootstrapFileStatus(); | ||
| FileSplit externalFileSplit = makeSplit(externalFileStatus.getPath(), 0, externalFileStatus.getLen(), | ||
| new String[0], new String[0]); | ||
| return new BootstrapBaseFileSplit(split, externalFileSplit); | ||
| } catch (IOException e) { | ||
| throw new HoodieIOException(e.getMessage(), e); | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -209,38 +225,6 @@ private List<FileStatus> listStatusForSnapshotModeLegacy(JobConf job, Map<String | |
| return HoodieInputFormatUtils.filterFileStatusForSnapshotMode(job, tableMetaClientMap, snapshotPaths, includeLogFilesForSnapshotView()); | ||
| } | ||
|
|
||
| @Nonnull | ||
| private List<FileStatus> listStatusForSnapshotMode(JobConf job, | ||
| Map<String, HoodieTableMetaClient> tableMetaClientMap, | ||
|
|
@@ -317,4 +301,80 @@ private List<FileStatus> listStatusForSnapshotMode(JobConf job, | |
|
|
||
| return targetFiles; | ||
| } | ||
|
|
||
| private void validate(List<FileStatus> targetFiles, List<FileStatus> legacyFileStatuses) { | ||
|
Contributor
There was a problem hiding this comment. Are the methods below merely moved? Looks like the order changes.
Contributor
Author
There was a problem hiding this comment. Yeah, GJF rule is applied (non-static methods) (static methods) |
||
| 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); | ||
| } | ||
| } | ||
|
|
||
| @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 static RealtimeFileStatus createRealtimeFileStatusUnchecked(HoodieLogFile latestLogFile, | ||
| Stream<HoodieLogFile> logFiles, | ||
| Option<HoodieInstant> latestCompletedInstantOpt, | ||
| HoodieTableMetaClient tableMetaClient) { | ||
| List<HoodieLogFile> sortedLogFiles = logFiles.sorted(HoodieLogFile.getLogFileComparator()).collect(Collectors.toList()); | ||
| try { | ||
| RealtimeFileStatus rtFileStatus = new RealtimeFileStatus(latestLogFile.getFileStatus()); | ||
| rtFileStatus.setDeltaLogFiles(sortedLogFiles); | ||
| rtFileStatus.setBasePath(tableMetaClient.getBasePath()); | ||
|
|
||
| if (latestCompletedInstantOpt.isPresent()) { | ||
| HoodieInstant latestCompletedInstant = latestCompletedInstantOpt.get(); | ||
| checkState(latestCompletedInstant.isCompleted()); | ||
|
|
||
| rtFileStatus.setMaxCommitTime(latestCompletedInstant.getTimestamp()); | ||
| } | ||
|
|
||
| return rtFileStatus; | ||
| } catch (IOException e) { | ||
| throw new HoodieIOException(String.format("Failed to init %s", RealtimeFileStatus.class.getSimpleName()), e); | ||
| } | ||
| } | ||
|
|
||
| private static Option<HoodieInstant> fromScala(scala.Option<HoodieInstant> opt) { | ||
| if (opt.isDefined()) { | ||
| return Option.of(opt.get()); | ||
| } | ||
|
|
||
| return Option.empty(); | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.