xiarixiaoyao commented on a change in pull request #3203:
URL: https://github.com/apache/hudi/pull/3203#discussion_r725781031
##########
File path:
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/HoodieParquetRealtimeInputFormat.java
##########
@@ -66,6 +91,139 @@
return HoodieRealtimeInputFormatUtils.getRealtimeSplits(job, fileSplits);
}
+ /**
+ * keep the logical of mor_incr_view as same as spark datasource.
+ * to do: unify the incremental view code between hive/spark-sql and spark
datasource
+ */
+ @Override
+ protected List<FileStatus> listStatusForIncrementalMode(
+ JobConf job, HoodieTableMetaClient tableMetaClient, List<Path>
inputPaths) throws IOException {
+ List<FileStatus> result = new ArrayList<>();
+ String tableName = tableMetaClient.getTableConfig().getTableName();
+ Job jobContext = Job.getInstance(job);
+
+ Option<HoodieTimeline> timeline =
HoodieInputFormatUtils.getFilteredCommitsTimeline(jobContext, tableMetaClient);
+ if (!timeline.isPresent()) {
+ return result;
+ }
+ String lastIncrementalTs = HoodieHiveUtils.readStartCommitTime(jobContext,
tableName);
+ // Total number of commits to return in this batch. Set this to -1 to get
all the commits.
+ Integer maxCommits = HoodieHiveUtils.readMaxCommits(jobContext, tableName);
+ HoodieTimeline commitsTimelineToReturn =
timeline.get().findInstantsAfter(lastIncrementalTs, maxCommits);
+ Option<List<HoodieInstant>> commitsToCheck =
Option.of(commitsTimelineToReturn.getInstants().collect(Collectors.toList()));
+ if (!commitsToCheck.isPresent()) {
+ return result;
+ }
+ Map<String, HashMap<String, FileStatus>> partitionsWithFileStatus =
HoodieInputFormatUtils
+ .listAffectedFilesForCommits(new Path(tableMetaClient.getBasePath()),
commitsToCheck.get(), commitsTimelineToReturn);
+ // build fileGroup from fsView
+ List<FileStatus> affectedFileStatus = new ArrayList<>();
+ partitionsWithFileStatus.forEach((key, value) -> value.forEach((k, v) ->
affectedFileStatus.add(v)));
+ HoodieTableFileSystemView fsView = new
HoodieTableFileSystemView(tableMetaClient, commitsTimelineToReturn,
affectedFileStatus.toArray(new FileStatus[0]));
+ // build fileGroup from fsView
+ String basePath = tableMetaClient.getBasePath();
+ // filter affectedPartition by inputPaths
+ List<String> affectedPartition = partitionsWithFileStatus.keySet().stream()
+ .filter(k -> k.isEmpty() ? inputPaths.contains(new Path(basePath)) :
inputPaths.contains(new Path(basePath, k))).collect(Collectors.toList());
+ if (affectedPartition.isEmpty()) {
+ return result;
+ }
+ List<HoodieFileGroup> fileGroups = affectedPartition.stream()
+ .flatMap(partitionPath ->
fsView.getAllFileGroups(partitionPath)).collect(Collectors.toList());
+ setInputPaths(job, affectedPartition.stream()
+ .map(p -> p.isEmpty() ? basePath : new Path(basePath,
p).toUri().toString()).collect(Collectors.joining(",")));
+
+ // find all file status in current partitionPath
+ FileStatus[] fileStatuses = getStatus(job);
+ Map<String, FileStatus> candidateFileStatus = new HashMap<>();
+ for (int i = 0; i < fileStatuses.length; i++) {
+ String key = fileStatuses[i].getPath().toString();
+ candidateFileStatus.put(key, fileStatuses[i]);
+ }
+
+ String maxCommitTime = fsView.getLastInstant().get().getTimestamp();
+ fileGroups.stream().forEach(f -> {
+ try {
+ List<FileSlice> baseFiles = f.getAllFileSlices().filter(slice ->
slice.getBaseFile().isPresent()).collect(Collectors.toList());
+ if (!baseFiles.isEmpty()) {
+ FileStatus baseFileStatus =
HoodieInputFormatUtils.getFileStatus(baseFiles.get(0).getBaseFile().get());
+ String baseFilePath = baseFileStatus.getPath().toUri().toString();
+ if (!candidateFileStatus.containsKey(baseFilePath)) {
+ throw new HoodieException("Error obtaining fileStatus for file: "
+ baseFilePath);
+ }
+ RealtimeFileStatus fileStatus = new
RealtimeFileStatus(candidateFileStatus.get(baseFilePath));
+ fileStatus.setMaxCommitTime(maxCommitTime);
+ fileStatus.setBelongToIncrementalFileStatus(true);
+ fileStatus.setBasePath(basePath);
+ fileStatus.setBaseFilePath(baseFilePath);
+
fileStatus.setDeltaLogPaths(f.getLatestFileSlice().get().getLogFiles().map(l ->
l.getPath().toString()).collect(Collectors.toList()));
Review comment:
// Ensure get the base file when there is a pending compaction,
which means the base file
// won't be in the latest file slice.
see the doc of MergeOnReadIncrementalRelation
https://github.com/apache/hudi/blob/ceace1c653a3ce3c97e6ee5a244d71ff1806be4f/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelation.scala#L203
--
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-unsubscr...@hudi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
