xiarixiaoyao commented on a change in pull request #3203:
URL: https://github.com/apache/hudi/pull/3203#discussion_r725743160
##########
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();
Review comment:
great suggest, i will split this method and add docs. thanks
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
