xiarixiaoyao commented on a change in pull request #3203:
URL: https://github.com/apache/hudi/pull/3203#discussion_r725737982
##########
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());
Review comment:
yes, we apply this pr in our product env, and find that bug.
--
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]
