[GitHub] [hudi] xiarixiaoyao commented on a change in pull request #3203: [HUDI-2086] Refactor hive mor_incremental_view

[GitBox <gi...@apache.org>]

xiarixiaoyao commented on a change in pull request #3203:
URL: https://github.com/apache/hudi/pull/3203#discussion_r740695703



##########
File path: hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieRealtimeInputFormatUtils.java
##########
@@ -161,6 +148,82 @@
     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) {
+        BaseFileWithLogsSplit bs = (BaseFileWithLogsSplit)s;
+        return bs.getBelongToIncrementalSplit();
+      } else {
+        return s instanceof RealtimeBootstrapBaseFileSplit;
+      }
+    });
+  }
+
+  public static RealtimeBootstrapBaseFileSplit createRealimeBootstrapBaseFileSplit(

Review comment:
       ok




-- 
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-unsubscribe@hudi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org