danny0405 commented on a change in pull request #3203:
URL: https://github.com/apache/hudi/pull/3203#discussion_r737998688
##########
File path:
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeInputFormatUtils.java
##########
@@ -161,6 +148,110 @@
return rtSplits.toArray(new InputSplit[0]);
}
+ // get IncrementalRealtimeSplits
+ public static InputSplit[] getIncrementalRealtimeSplits(Configuration conf,
Stream<FileSplit> fileSplits) throws IOException {
+ List<InputSplit> rtSplits = new ArrayList<>();
+ List<FileSplit> fileSplitList = fileSplits.collect(Collectors.toList());
+ Set<Path> partitionSet = fileSplitList.stream().map(f ->
f.getPath().getParent()).collect(Collectors.toSet());
+ Map<Path, HoodieTableMetaClient> partitionsToMetaClient =
getTableMetaClientByPartitionPath(conf, partitionSet);
+ // Pre process tableConfig from first partition to fetch virtual key info
+ Option<HoodieVirtualKeyInfo> hoodieVirtualKeyInfo = Option.empty();
+ if (partitionSet.size() > 0) {
+ hoodieVirtualKeyInfo =
getHoodieVirtualKeyInfo(partitionsToMetaClient.get(partitionSet.iterator().next()));
+ }
+ Option<HoodieVirtualKeyInfo> finalHoodieVirtualKeyInfo =
hoodieVirtualKeyInfo;
+ fileSplitList.stream().forEach(s -> {
+ // deal with incremental query.
+ try {
+ if (s instanceof BaseFileWithLogsSplit) {
+ BaseFileWithLogsSplit bs = (BaseFileWithLogsSplit)s;
+ if (bs.getBelongToIncrementalSplit()) {
+ rtSplits.add(new HoodieRealtimeFileSplit(bs, bs.getBasePath(),
bs.getDeltaLogPaths(), bs.getMaxCommitTime(), finalHoodieVirtualKeyInfo));
+ }
+ } else if (s instanceof RealtimeBootstrapBaseFileSplit) {
+ rtSplits.add(s);
+ }
+ } catch (IOException e) {
+ throw new HoodieIOException("Error creating hoodie real time split ",
e);
+ }
+ });
+ LOG.info("Returning a total splits of " + rtSplits.size());
+ return rtSplits.toArray(new InputSplit[0]);
+ }
+
+ public static Option<HoodieVirtualKeyInfo>
getHoodieVirtualKeyInfo(HoodieTableMetaClient metaClient) {
+ HoodieTableConfig tableConfig = metaClient.getTableConfig();
+ if (!tableConfig.populateMetaFields()) {
+ TableSchemaResolver tableSchemaResolver = new
TableSchemaResolver(metaClient);
+ try {
+ MessageType parquetSchema =
tableSchemaResolver.getTableParquetSchema();
+ return Option.of(new
HoodieVirtualKeyInfo(tableConfig.getRecordKeyFieldProp(),
+ tableConfig.getPartitionFieldProp(),
parquetSchema.getFieldIndex(tableConfig.getRecordKeyFieldProp()),
+ parquetSchema.getFieldIndex(tableConfig.getPartitionFieldProp())));
+ } catch (Exception exception) {
+ throw new HoodieException("Fetching table schema failed with exception
", exception);
+ }
+ }
+ return Option.empty();
+ }
+
+ public static boolean isIncrementalQuerySplits(List<FileSplit> fileSplits) {
+ if (fileSplits == null || fileSplits.size() == 0) {
+ return false;
+ }
+ return fileSplits.stream().anyMatch(s -> {
+ if (s instanceof BaseFileWithLogsSplit) {
Review comment:
Can be simplified as:
```java
return fileSplits.stream().anyMatch(s -> {
if (s instanceof BaseFileWithLogsSplit) {
BaseFileWithLogsSplit bs = (BaseFileWithLogsSplit)s;
return bs.getBelongToIncrementalSplit();
} else {
return s instanceof RealtimeBootstrapBaseFileSplit;
}
});
```
--
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
